Рассмотрен примитивный метод поиска похожих графиков с помощью корреляции. Все происходит под Linux с помощью Python 3.5. (Windows может добавить геморроя.)
Основная идея: когда нравится движение цены на графике в определенный момент времени, я хочу легко находить похожие движения на рынке на сегодняшний день.
Исходные данные:
- Linux (Ubuntu).
- Python 3.5.
- История цен.
- Ipython (Jupyter).
Цены можно выкачивать с Yahoo.Finance, но это будет крайне медленно. Отпимальным решением будет бесплатная база на Quandl.
🎓Корреляция: что это?
Объяснение и формулу можно почерпнуть на Вики, а кратко: это статистическая взаимосвязь двух или более случайных величин.
В данном случае мы будем использовать корреляцию Пирсона. С помощью данного вида корреляции можно определить силу линейной зависимости между величинами. Корреляция будет измеряться от -1 (обратная корреляция, цены движутся в противоположных направлениях) до 1 (прямая корреляция, цены движутся в одном направлении).
Искать корреляцию будем с помощью библиотеки talib, так как она позволяет искать максимально быстро из доступных подручных пакетов для Python. Как устанавливать библиотеку можно почитать здесь. Расширение для Python устанавливаем командой:
1 |
$ pip install TA-Lib |
Пример использования:
1 2 3 4 5 6 7 |
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.
Примеры использования:
1 2 3 4 5 6 7 8 9 |
# 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) |
Тестируем скорость (для пытливых умов):
1 2 3 4 5 |
%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.
Есть погрешности, графики не идеально похожи, однако общая тенденция видна, и это было найдено за несколько секунд.
Дополнительно можно создавать свои штампы, например:
Ищем ровные тренды
1 |
[x/2 for x in range(0, 100, 1)] |
Ищем параболы
1 |
[x*x for x in range(-10, 11, 1)] |
🎁Код в студию
Блокнот Ipython с кодом доступен в репозитории. Там достаточно поменять настройки подключения к базе данных и sql-запрос получения цен. У меня цены хранятся в integer и при получении их необходимо поделить на 10000.
Подключение к базе данных
Здесь описан класс для работы с базой данных. При создании передаем параметры соединения. Подтверждением успеха будет тишина…
Далее по коду будем дергать метод [code python]Db().query()[/code] для выполнения запросов и получения данных.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 |
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() — поиск корреляций для списка тикеров и возвращение тикеров с корреляцией выше/ниже указанной границы.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
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 |
Готовим штамп для поиска
За основу можно взять интересующий отрезок истории изменения цены существующего актива или генерировать значения по формуле. Дополнительно надо указать количество анализируемых сессий и порог корреляции.
1 2 3 4 5 6 7 8 9 10 11 12 |
# 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] и результатом получаем найденные тикеры, отсортированные в сторону большей корреляции.
1 2 3 4 5 6 7 8 9 |
# 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.