Поиск графиков по шаблону через корреляцию

Рассмотрен примитивный метод поиска похожих графиков с помощью корреляции. Все происходит под Linux с помощью Python 3.5. (Windows может добавить геморроя.)

Основная идея: когда нравится движение цены на графике в определенный момент времени, я хочу легко находить похожие движения на рынке на сегодняшний день.

Исходные данные:

Linux (Ubuntu).
Python 3.5.
История цен.
Ipython (Jupyter).

Цены можно выкачивать с Yahoo.Finance, но это будет крайне медленно. Отпимальным решением будет бесплатная база на Quandl.

🎓Корреляция: что это?

Объяснение и формулу можно почерпнуть на Вики, а кратко: это статистическая взаимосвязь двух или более случайных величин.

В данном случае мы будем использовать корреляцию Пирсона. С помощью данного вида корреляции можно определить силу линейной зависимости между величинами. Корреляция будет измеряться от -1 (обратная корреляция, цены движутся в противоположных направлениях) до 1 (прямая корреляция, цены движутся в одном направлении).

Искать корреляцию будем с помощью библиотеки talib, так как она позволяет искать максимально быстро из доступных подручных пакетов для Python. Как устанавливать библиотеку можно почитать здесь. Расширение для Python устанавливаем командой:

$ pip install TA-Lib

1	$ pip install TA-Lib

Пример использования:

import numpy as np
import talib

arr = np.array([x for x in range(0, 100, 1)], dtype=float)
length = 50 # length of data for get correlation
corr = talib.CORREL(arr[-length:], arr[-length:], length)
print(corr[-1])

import numpy as np

import talib

arr = np.array([x for x in range(0, 100, 1)], dtype=float)

length = 50 # length of data for get correlation

corr = talib.CORREL(arr[-length:], arr[-length:], length)

print(corr[-1])

В данном примере мы получим корреляцию для 50 последовательных элементов. Результатом будет единица, так как сравниваемая выборка равна самой себе.

Есть альтернативы — numpy.corrcoef и scipy.stats.pearsonr. Работают чуть медленнее, но значительно легче устанавливаются. Scipy.stats.pearsonr работает быстрее реализации numpy.

Примеры использования:

# SciPy & Numpy
import numpy as np
from scipy.stats import pearsonr

arr = np.array([x for x in range(0, 100, 1)], dtype=float)
length = 50 # length of data for get correlation
corr_np = np.corrcoef(arr[-length:], arr[-length:])
corr_scipy = pearsonr(arr[-length:], arr[-length:])
print(corr_np, corr_scipy)

# SciPy & Numpy

import numpy as np

from scipy.stats import pearsonr

arr = np.array([x for x in range(0, 100, 1)], dtype=float)

length = 50 # length of data for get correlation

corr_np = np.corrcoef(arr[-length:], arr[-length:])

corr_scipy = pearsonr(arr[-length:], arr[-length:])

print(corr_np, corr_scipy)

Тестируем скорость (для пытливых умов):

%timeit talib.CORREL(arr[-length:], arr[-length:], length)
%timeit np.corrcoef(arr[-length:], arr[-length:])
%timeit pearsonr(arr[-length:], arr[-length:])

%timeit talib.CORREL(arr[-length:], arr[-length:], length)

%timeit np.corrcoef(arr[-length:], arr[-length:])

%timeit pearsonr(arr[-length:], arr[-length:])

🔭Где взять историю цен?

Для ускорения поиска похожих графиков историю цен лучше иметь локально в базе данных. Можно выкачать все активы с Yahoo.Finance, но это и сложно и долго. Оптимальным решением вижу Quandl, где бесплатно доступны 3000 активов совершенно бесплатно, что будет более чем достаточно, чтобы поиграться. Примеры кода есть на Quandl.

Ищем шаблоны цены

Мне нравятся шаблоны следующих графиков:

CIM — 30 сессий до 25 июля 2016

WIN — 20 сессий до 5 сентября 2016

Чтобы найти подобные графики нам нужен штамп изменения цены за определенное время. Получаем историю изменения цены (используем только adjusted close) и заворачиваем в numpy-массив.

Далее делаем запрос и получаем историю цен всех активов на заданную продолжительность. Для каждого актива заворачиваем close в numpy-массив аналогично штампу.

Итогом у нас имеется массив со штампом и список массивов с ценами для сравнения.

Вот результат на 11 октября 2016 по шаблону CIM (PTCT, KERX, CARA, LITE):

Сравниваем 30 торговых сессий с порогом корреляции выше 0,93.

Результат на 11 октября 2016 по шаблону WIN (TLDR, MHLD):

Сравниваем 20 торговых сессий с порогом корреляции выше 0,85.

Есть погрешности, графики не идеально похожи, однако общая тенденция видна, и это было найдено за несколько секунд.

Дополнительно можно создавать свои штампы, например:

Ищем ровные тренды

[x/2 for x in range(0, 100, 1)]

1	[x/2 for x in range(0, 100, 1)]

Ищем параболы

[x*x for x in range(-10, 11, 1)]

1	[x*x for x in range(-10, 11, 1)]

🎁Код в студию

Блокнот Ipython с кодом доступен в репозитории. Там достаточно поменять настройки подключения к базе данных и sql-запрос получения цен. У меня цены хранятся в integer и при получении их необходимо поделить на 10000.

Подключение к базе данных

Здесь описан класс для работы с базой данных. При создании передаем параметры соединения. Подтверждением успеха будет тишина…

Далее по коду будем дергать метод [code python]Db().query()[/code] для выполнения запросов и получения данных.

import psycopg2
import psycopg2.extras

class Db(object):
    _connection = None
    
    def __init__(self, host="localhost", user="user", password="1", db="db"):
        try:
            self._connection = psycopg2.connect("host='{0}' user='{1}' password='{2}' dbname='{3}'".format(
                    host, user, password, db))
        except psycopg2.Error as err:
            print("Connection error: {}".format(err))
            self._connection.close()
    
    def query(self, sql, params=None, cursor='list'):
        if not self._connection:
            return False
        
        data = False
        
        if cursor == 'dict':
            # Assoc cursor
            factory = psycopg2.extras.DictCursor
        else:
            # Standard cursor
            factory = psycopg2.extensions.cursor
        try:
            cur = self._connection.cursor(cursor_factory=factory) # by column name
            cur.execute(sql, params)
            data = cur.fetchall()
        except psycopg2.Error as err:
            print("Query error: {}".format(err))
            
        return data

db = Db(host="localhost", user="developer", password="1", db="test_database")

import psycopg2

import psycopg2.extras

class Db(object):

_connection = None

def __init__(self, host="localhost", user="user", password="1", db="db"):

try:

self._connection = psycopg2.connect("host='{0}' user='{1}' password='{2}' dbname='{3}'".format(

host, user, password, db))

except psycopg2.Error as err:

print("Connection error: {}".format(err))

self._connection.close()

def query(self, sql, params=None, cursor='list'):

if not self._connection:

return False

data = False

if cursor == 'dict':

# Assoc cursor

factory = psycopg2.extras.DictCursor

else:

# Standard cursor

factory = psycopg2.extensions.cursor

try:

cur = self._connection.cursor(cursor_factory=factory) # by column name

cur.execute(sql, params)

data = cur.fetchall()

except psycopg2.Error as err:

print("Query error: {}".format(err))

return data

db = Db(host="localhost", user="developer", password="1", db="test_database")

Вспомогательные функции

prices() — получение истории цен на заданный промежуток времени для массива тикеров.
check_stamp() — поиск корреляций для списка тикеров и возвращение тикеров с корреляцией выше/ниже указанной границы.

import operator
import numpy as np
import talib
from datetime import date, timedelta
table = 'prices'
order_dt = 'ASC'

def prices(symbols, dt_from = date.today(), period = 90):
    """
    Get prices from database for one or multiple symbols.
    """
    dt_to = dt_from - timedelta(days=period)
    cond = {
        'symbols': symbols,
        'to': dt_from,
        'from': dt_to
    }

    sql = """
        SELECT
            symbol,
            dt,
            ROUND(CASE WHEN adj IS NOT NULL AND close > 0
                THEN adj / close::real ELSE 1 END * open) / {0} as open,
            ROUND(CASE WHEN adj IS NOT NULL AND close > 0
                THEN adj / close::real ELSE 1 END * high) / {0} as high,
            ROUND(CASE WHEN adj IS NOT NULL AND close > 0
                THEN adj / close::real ELSE 1 END * low) / {0} as low,
            CASE WHEN adj IS NOT NULL THEN adj ELSE close
                END::real / {0} as close,
            volume::int
        FROM {1}
        WHERE symbol IN (SELECT unnest(%(symbols)s)) 
            AND dt BETWEEN %(from)s AND %(to)s
        ORDER BY dt {2}""".format(10000, table, order_dt)

    return db.query(sql, cond)

def check_stamp(symbols, stamp, length=30, corr=0.9):
    """
    Get prices for list of symbols and get correlation
    for every prices history.
    Leave in result only more correlated symbols.
    """
    to_check = prices(symbols, date.today(), 200)
    
    stamp_change = (stamp[1:] - stamp[:-1])
    stamp_change_prc = stamp_change / stamp[:-1]

    to_check_price = dict()
    corr_price = {s:0 for s in symbols}
    
    for symbol in symbols:
        to_check_price[symbol] = np.array([x[5] for x in to_check if x[0] == symbol])    
        if to_check_price[symbol].shape[0] < length: 
            print("Symbol {0} does not have enough history ({1}).".format(symbol, to_check_price[symbol].shape[0]))
            continue
        corr_price[symbol] = talib.CORREL(stamp[-length:], to_check_price[symbol][-length:], length)[-1] 
    
    sorted_price = sorted(corr_price.items(), key=operator.itemgetter(1))
    # filter by correlation
    sorted_price = ([r for r in sorted_price if r[1] >= corr] if corr > 0 
                    else [r for r in sorted_price if r[1] <= corr]) 
    return sorted_price

import operator

import numpy as np

import talib

from datetime import date, timedelta

table = 'prices'

order_dt = 'ASC'

def prices(symbols, dt_from = date.today(), period = 90):

"""

Get prices from database for one or multiple symbols.

"""

dt_to = dt_from - timedelta(days=period)

cond = {

'symbols': symbols,

'to': dt_from,

'from': dt_to

}

sql = """

SELECT

symbol,

dt,

ROUND(CASE WHEN adj IS NOT NULL AND close > 0

THEN adj / close::real ELSE 1 END * open) / {0} as open,

ROUND(CASE WHEN adj IS NOT NULL AND close > 0

THEN adj / close::real ELSE 1 END * high) / {0} as high,

ROUND(CASE WHEN adj IS NOT NULL AND close > 0

THEN adj / close::real ELSE 1 END * low) / {0} as low,

CASE WHEN adj IS NOT NULL THEN adj ELSE close

END::real / {0} as close,

volume::int

FROM {1}

WHERE symbol IN (SELECT unnest(%(symbols)s))

AND dt BETWEEN %(from)s AND %(to)s

ORDER BY dt {2}""".format(10000, table, order_dt)

return db.query(sql, cond)

def check_stamp(symbols, stamp, length=30, corr=0.9):

"""

Get prices for list of symbols and get correlation

for every prices history.

Leave in result only more correlated symbols.

"""

to_check = prices(symbols, date.today(), 200)

stamp_change = (stamp[1:] - stamp[:-1])

stamp_change_prc = stamp_change / stamp[:-1]

to_check_price = dict()

corr_price = {s:0 for s in symbols}

for symbol in symbols:

to_check_price[symbol] = np.array([x[5] for x in to_check if x[0] == symbol])

if to_check_price[symbol].shape[0] < length:

print("Symbol {0} does not have enough history ({1}).".format(symbol, to_check_price[symbol].shape[0]))

continue

corr_price[symbol] = talib.CORREL(stamp[-length:], to_check_price[symbol][-length:], length)[-1]

sorted_price = sorted(corr_price.items(), key=operator.itemgetter(1))

# filter by correlation

sorted_price = ([r for r in sorted_price if r[1] >= corr] if corr > 0

else [r for r in sorted_price if r[1] <= corr])

return sorted_price

Готовим штамп для поиска

За основу можно взять интересующий отрезок истории изменения цены существующего актива или генерировать значения по формуле. Дополнительно надо указать количество анализируемых сессий и порог корреляции.

# stamp by price history
stamp = prices(['CIM'], date(2016, 7, 25), 200) # length=30, corr=0.93, long
# stamp = prices(['WIN'], date(2016, 9, 5), 200) # length=20, 0.90, long # inverted parabola
# stamp = np.array([x[5] for x in stamp])

# stamp by function
# stamp = np.array([x/2 for x in range(0, 100, 1)], dtype=float) # length=30, corr=0.95, long # trend
# stamp = np.array([pow(1.05, i) for i in range(100)], dtype=float) # length=30, corr=0.95, long # growth trend
# stamp = np.array([x*x for x in range(-10, 11, 1)], dtype=float) # length=20, corr=0.90, long # inverted parabola

length, corr = 30, 0.93
stamp

# stamp by price history

stamp = prices(['CIM'], date(2016, 7, 25), 200) # length=30, corr=0.93, long

# stamp = prices(['WIN'], date(2016, 9, 5), 200) # length=20, 0.90, long # inverted parabola

# stamp = np.array([x[5] for x in stamp])

# stamp by function

# stamp = np.array([x/2 for x in range(0, 100, 1)], dtype=float) # length=30, corr=0.95, long # trend

# stamp = np.array([pow(1.05, i) for i in range(100)], dtype=float) # length=30, corr=0.95, long # growth trend

# stamp = np.array([x*x for x in range(-10, 11, 1)], dtype=float) # length=20, corr=0.90, long # inverted parabola

length, corr = 30, 0.93

stamp

🍰А теперь самое вкусное

Готовим список тикеров, которые будем проверять, или заполняем его вручную. Список тикеров со штампом и настройками кидаем на обработку в [code]check_stamp()[/code] и результатом получаем найденные тикеры, отсортированные в сторону большей корреляции.

# get active symbols
sql = "SELECT symbol FROM listed WHERE is_listed AND price > 1 AND avg_volume > 500000 ORDER BY symbol"
r = db.query(sql)
listed_symbols = [s[0] for s in r]

# listed_symbols = ['NVDA', 'SLV', 'MMM', 'SPY', 'DIA', 'QQQ', 'GLD', 'UUP']
                             
corr = check_stamp(listed_symbols, stamp, length, corr)
corr, ",".join([s[0] for s in corr])

# get active symbols

sql = "SELECT symbol FROM listed WHERE is_listed AND price > 1 AND avg_volume > 500000 ORDER BY symbol"

r = db.query(sql)

listed_symbols = [s[0] for s in r]

# listed_symbols = ['NVDA', 'SLV', 'MMM', 'SPY', 'DIA', 'QQQ', 'GLD', 'UUP']

corr = check_stamp(listed_symbols, stamp, length, corr)

corr, ",".join([s[0] for s in corr])

Графики найденных тикеров всем скопом можно посмотреть на finviz.

💬Напишите в комментариях, допустима ли такая погрешность? И надо ли так искать?

Александр Румянцев
Автор на Quantrum.me
Telegram-канал📣: @quantiki

Подбор и тестирование портфелей. Подключение роботов к IB.