Выбор акций с использованием стратегии количественного импульса

Знакомство

Когда я был трейдером-любителем, процесс выбора правильных акций для торговли был кошмаром. Новости об акциях, неопределенность и эмоции добавляют горечи этому процессу. Забегая далеко вперед, сегодня, я нашел свое собственное решение, используя своего лучшего компаньона Python. В этой статье мы собираемся построить простую количественную стратегию импульса на python, которая фильтрует и выбирает лучшие внутридневные акции. Но подождите, что такое количественная стратегия импульса?

Стратегия количественного импульса — это стратегия, реализуемая для выбора акций, которые больше всего выросли в цене. Проще говоря, это процесс идентификации акций с большим восходящим трендом. Теперь, когда мы получили некоторое представление о том, что такое стратегия количественного импульса и как ее можно использовать для выбора акций. Давайте реализуем стратегию на python!

Реализация Python

В этой статье рассматриваются следующие шаги:

- Importing the required packages- Extracting the list of all S&P 500 stock's symbols- Pulling Intraday data of all the stocks in the S&P 500- Calculating percentage change and momentum of all stocks- Finding stocks with greater momentum- Backtesting with a equal-weight portfolio

Шаг 1: Импорт пакетов

Основными пакетами этой статьи являются пакет Pandas для работы с данными, пакет NumPy для работы с массивами и пакет Requests для извлечения данных с помощью API. Вторичными пакетами будут пакет Math для математических функций, пакет SciPy для сложных функций и пакет Statistics для статистических функций.

Реализация Python:

import pandas as pd
import requests
import numpy as np
from scipy.stats import percentileofscore as score
from statistics import mean
from math import floor

Теперь, когда мы импортировали все необходимые пакеты в нашу среду Python. Давайте приступим к извлечению списка всех символов акций S&P 500.

Шаг 2: Извлечение списка всех символов акций S&P 500

Я видел учебные пособия, создающие сложные функции веб-скрейпинга, чтобы выполнить эту конкретную задачу. Но это не тот путь, по которому мы собираемся идти, и наш код будет намного проще и эффективнее.

Реализация Python:

sp500 = pd.read_html('https://en.wikipedia.org/wiki/List_of_S%26P_500_companies')
sp500_list = np.array(sp500[0]['Symbol'])
sp500_list[:20]

Пояснение к коду: Как я уже говорил ранее, мы не собираемся усложнять наш код или использовать какие-либо методы веб-скрейпинга для извлечения символов акций в S&P 500, но мы использовали высокоэффективную функцию под названием «read_html», предоставляемую пакетом Pandas. Эта функция будет искать таблицы по заданному URL-адресу и вернет данные в формате списка. Возвращаясь к нашему коду, мы сначала извлекли символы акций в S&P 500 с помощью функции «read_html» и сохранили их в переменной «sp500». Затем мы определили переменную ‘sp500_list’ для хранения извлеченных данных в массиве NumPy. Теперь, когда мы собрали все символы акций S&P 500. Давайте вытащим некоторые внутридневные данные!

Шаг 3: Извлечение внутридневных данных с помощью IEX Cloud API

На этом этапе мы собираемся получить внутридневные данные по всем акциям S&P 500 с помощью IEX Cloud API. Если вы не знаете, как извлекать биржевые данные с помощью IEX Cloud API, я предлагаю вам прочитать мою статью об этом здесь.

Реализация Python:

def get_intraday_prices(symbol):
    ticker = symbol
    iex_api_key = 'YOUR API KEY'
    url = f'https://cloud.iexapis.com/stable/stock/{ticker}/intraday-prices?token={iex_api_key}'
    df = requests.get(url).json()
    date = df[1]['date']
        
    time = []
    open = []
    high = []
    low = []
    close = []
    volume = []
    number_of_trades = []
    
    for i in range(len(df)):
        time.append(df[i]['label'])
        open.append(df[i]['open'])
        high.append(df[i]['high'])
        low.append(df[i]['low'])
        close.append(df[i]['close'])
        volume.append(df[i]['volume'])
        number_of_trades.append(df[i]['numberOfTrades'])
        
    time_df = pd.DataFrame(time).rename(columns = {0:'Time'})
    open_df = pd.DataFrame(open).rename(columns = {0:'Open'})
    high_df = pd.DataFrame(high).rename(columns = {0:'High'})
    low_df = pd.DataFrame(low).rename(columns = {0:'Low'})
    close_df = pd.DataFrame(close).rename(columns = {0:'Close'})
    volume_df = pd.DataFrame(volume).rename(columns = {0:'Volume'})
    number_of_trades_df = pd.DataFrame(number_of_trades).rename(columns = {0:'Number of Trades'})
     
    frames = [time_df, open_df, high_df, low_df, close_df, volume_df, number_of_trades_df]
    df = pd.concat(frames, axis = 1, join = 'inner')
    df = df.set_index('Time')
    return df
  
df = pd.DataFrame(columns = sp500_list)
for i in df.columns:
    try:
        df[i] = get_intraday_prices(i)['Close']
        print(f'{i} is successfully extracted')
    except:
        pass
    
df.to_csv('sp500.csv')
sp500 = pd.read_csv('sp500.csv').set_index('Time')
sp500.head()

Пояснение к коду: Этот фрагмент кода можно разделить на три части: определение функции для извлечения внутридневных данных, извлечение внутридневных данных по всем акциям в S&P 500 с использованием определенной функции, сохранение и импорт извлеченных данных.

Первая часть содержит код со строк с 1 по 36. В первой части мы сначала определяем функцию с именем ‘get_intraday_prices’, которая принимает символ акции в качестве параметра. Внутри функции мы сначала определяем две переменные для хранения ключа API и URL-адреса. Используя функцию get, предоставляемую пакетом Requests, мы извлекаем внутридневные данные в формате JSON. После некоторой обработки данных и манипуляций мы возвращаем данные в виде кадра данных Pandas.

Вторая часть содержит код из строк с 38 по 44. В этой части мы перебираем символы акций, чтобы получить внутридневные данные по каждой акции. Мы храним внутридневные данные каждой акции в переменной ‘df’.

Третья часть содержит код со строк 46 по 48. Основной целью этой части является сохранение и импорт извлеченных внутридневных данных по всем запасам. Этот шаг необязателен, но настоятельно рекомендуется, так как вы можете сэкономить время на запуске процесса итерации при повторном открытии скрипта. Сначала мы сохраняем извлеченные внутридневные данные от имени «sp500» с помощью функции «to_csv», предоставляемой пакетом Pandas. Затем с помощью функции «read_csv» мы импортируем сохраненный кадр данных.

Шаг 4: Расчет процентного изменения и импульса

На этом этапе мы собираемся рассчитать дневное изменение и импульс каждой акции.

Реализация Python:

# CALCULATING DAY CHANGE OF STOCKS

dc = []
for i in sp500.columns:
    dc.append(sp500[i].pct_change().sum())
    
sp500_momentum = pd.DataFrame(columns = ['symbol', 'day_change'])
sp500_momentum['symbol'] = sp500.columns
sp500_momentum['day_change'] = dc

# CALCULATING MOMENTUM

sp500_momentum['momentum'] = 'N/A'
for i in range(len(sp500_momentum)):
    sp500_momentum.loc[i, 'momentum'] = score(sp500_momentum.day_change, sp500_momentum.loc[i, 'day_change'])/100
    
sp500_momentum['momentum'] = sp500_momentum['momentum'].astype(float)    
sp500_momentum.head()

Пояснение к коду: Во-первых, мы создаем пустой список с именем «pc» для хранения процентного изменения каждой акции. Используя цикл for, мы перебираем символы всех акций в S&P 500 и добавляем рассчитанное процентное изменение с помощью функции «pct_change», предоставляемой пакетом Pandas, в список «dc». Затем мы создаем кадр данных с именем «sp500_momentum» для хранения символа акции, процентного изменения и импульса, который мы собираемся вычислить.

Переходя к коду для вычисления импульса, мы сначала создаем столбец с именем «momentum» в кадре данных «sp500_momentum» и заполняем его нулевыми значениями. Затем мы передаем цикл for, чтобы заполнить нулевые значения фактическими значениями импульса.

Шаг 5: Поиск акций, имеющих больший импульс

На этом этапе мы собираемся отсортировать акции на основе рассчитанного нами значения импульса и найти 10 лучших акций, имеющих наибольший импульс.

Реализация Python:

top_picks = sp500_momentum.nlargest(10, 'momentum')['symbol'].reset_index().drop('index', axis = 1)
top_picks

Пояснение к коду: Я всегда следую принципу простоты и эффективности кода. Я делал это и здесь. Сначала мы создаем фрейм данных с именем «top_picks», в котором хранятся 10 крупнейших акций, имеющих больший импульс, чем остальные. Чтобы отсортировать и найти топ-10 акций, мы использовали функцию «nlarge», предоставляемую пакетом Pandas.

Шаг 6: Тестирование на истории

На этом этапе мы собираемся провести тестирование, вложив равную сумму денег в топ-10 акций с наибольшим импульсом, и давайте посмотрим на результаты.

Реализация Python:

portfolio_val = 1000000
per_stock_val = portfolio_val/len(top_picks)

day_close = []
for i in top_picks['symbol']:
    data = sp500[i]
    day_close.append(data[-1])
    
backtest_df = pd.DataFrame(columns = ['selected_symbols', 'day_close', 'number_of_stocks', 'return', 'return_percentage'])
backtest_df['selected_symbols'] = top_picks['symbol']
backtest_df['day_close'] = day_close
for i in range(len(backtest_df)):
    backtest_df.loc[i, 'number_of_stocks'] = floor(per_stock_val/day_close[i])
    
returns = []
for i in top_picks['symbol']:
    ret = np.diff(sp500[i])
    ret = ret[~np.isnan(ret)]
    returns.append(round(sum(ret), 2))
    
backtest_returns = []
return_percentage = []
for i in range(len(backtest_df)):
    br = returns[i]*backtest_df.loc[i, 'number_of_stocks']
    rp = br/per_stock_val*100
    backtest_returns.append(round(br, 2))
    return_percentage.append(round(rp, 2))
backtest_df['return'] = backtest_returns
backtest_df['return_percentage'] = return_percentage

backtest_df

Пояснение к коду: Во-первых, мы определяем две переменные для хранения общей стоимости инвестиций и стоимости инвестиций на акцию соответственно. Затем мы проходим цикл for, чтобы найти дневную цену закрытия всех 10 крупнейших акций и добавить эти значения в переменную «day_close». В дальнейших строках кода мы создали кадр данных с именем «backtest_df», который включает в себя рассчитанное количество акций для покупки в каждой из них, доходность, которую мы получили, инвестируя в эти 10 лучших акций, и, наконец, процент доходности наших инвестиций. Как видите, мы получили приятную отдачу. Ну вот!

Заключительные мысли!

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

  • Выбор акций для долгосрочного инвестирования: В этой статье мы выбрали акции для внутридневной торговли, но вы также можете реализовать стратегию выбора долгосрочных инвестиционных акций. Делая это, вы узнаете, как соответствующим образом настроить стратегию количественного импульса и использовать несколько показателей для построения стратегии.
  • Больше акций: Вся эта статья основана только на акциях, перечисленных в рыночном индексе S&P 500, но вы можете рассмотреть акции, котирующиеся на многих других биржах. Преимущество выполнения этой задачи заключается в том, что вы сможете проводить исследования в больших масштабах, учитывая объем извлекаемых данных, и вы можете овладеть искусством извлечения данных с помощью API акций наряду с обработкой данных, поскольку эта задача имеет большое количество.

Ну вот! Вы дошли до конца этой статьи. Кроме того, если вы забыли следовать какой-либо из частей кодирования, не волнуйтесь! Я предоставил полный исходный код в конце. Надеюсь, вы нашли что-то полезное в этой статье. Если у вас есть какие-либо предложения по улучшению этой статьи, пожалуйста, оставьте комментарий.

Полный код:

import pandas as pd
import requests
import numpy as np
from scipy.stats import percentileofscore as score
from statistics import mean
from math import floor

# CREATING S&P 500 SYMBOL LIST

sp500 = pd.read_html('https://en.wikipedia.org/wiki/List_of_S%26P_500_companies')
sp500_list = np.array(sp500[0]['Symbol'])
print(sp500_list[:20])

# EXTRACTING INTRADAY PRICES OF S&P 500 STOCKS

def get_intraday_prices(symbol):
    ticker = symbol
    iex_api_key = 'pk_32ef64b2003542b6a829b8f94831c789'
    url = f'https://cloud.iexapis.com/stable/stock/{ticker}/intraday-prices?token={iex_api_key}'
    df = requests.get(url).json()
    date = df[1]['date']
        
    time = []
    open = []
    high = []
    low = []
    close = []
    volume = []
    number_of_trades = []
    
    for i in range(len(df)):
        time.append(df[i]['label'])
        open.append(df[i]['open'])
        high.append(df[i]['high'])
        low.append(df[i]['low'])
        close.append(df[i]['close'])
        volume.append(df[i]['volume'])
        number_of_trades.append(df[i]['numberOfTrades'])
        
    time_df = pd.DataFrame(time).rename(columns = {0:'Time'})
    open_df = pd.DataFrame(open).rename(columns = {0:'Open'})
    high_df = pd.DataFrame(high).rename(columns = {0:'High'})
    low_df = pd.DataFrame(low).rename(columns = {0:'Low'})
    close_df = pd.DataFrame(close).rename(columns = {0:'Close'})
    volume_df = pd.DataFrame(volume).rename(columns = {0:'Volume'})
    number_of_trades_df = pd.DataFrame(number_of_trades).rename(columns = {0:'Number of Trades'})
     
    frames = [time_df, open_df, high_df, low_df, close_df, volume_df, number_of_trades_df]
    df = pd.concat(frames, axis = 1, join = 'inner')
    df = df.set_index('Time')
    return df
  
  df = pd.DataFrame(columns = sp500_list)
for i in df.columns:
    try:
        df[i] = get_intraday_prices(i)['Close']
        print(f'{i} is successfully extracted')
    except:
        pass
    
df.to_csv('sp500.csv')

# IMPORTING THE EXTRACTED INTRADAY DATA

sp500 = pd.read_csv('sp500.csv').set_index('Time')
print(sp500.head())

# CALCULATING DAY CHANGE OF STOCKS

dc = []
for i in sp500.columns:
    dc.append(sp500[i].pct_change().sum())
    
sp500_momentum = pd.DataFrame(columns = ['symbol', 'day_change'])
sp500_momentum['symbol'] = sp500.columns
sp500_momentum['day_change'] = dc

# CALCULATING MOMENTUM

sp500_momentum['momentum'] = 'N/A'
for i in range(len(sp500_momentum)):
    sp500_momentum.loc[i, 'momentum'] = score(sp500_momentum.day_change, sp500_momentum.loc[i, 'day_change'])/100
    
sp500_momentum['momentum'] = sp500_momentum['momentum'].astype(float)    
print(sp500_momentum.head())

top_picks = sp500_momentum.nlargest(10, 'momentum')['symbol'].reset_index().drop('index', axis = 1)
print(top_picks)

# BACKTEST

portfolio_val = 1000000
per_stock_val = portfolio_val/len(top_picks)

day_close = []
for i in top_picks['symbol']:
    data = sp500[i]
    day_close.append(data[-1])
    
backtest_df = pd.DataFrame(columns = ['selected_symbols', 'day_close', 'number_of_stocks', 'return', 'return_percentage'])
backtest_df['selected_symbols'] = top_picks['symbol']
backtest_df['day_close'] = day_close
for i in range(len(backtest_df)):
    backtest_df.loc[i, 'number_of_stocks'] = floor(per_stock_val/day_close[i])
    
returns = []
for i in top_picks['symbol']:
    ret = np.diff(sp500[i])
    ret = ret[~np.isnan(ret)]
    returns.append(round(sum(ret), 2))
    
backtest_returns = []
return_percentage = []
for i in range(len(backtest_df)):
    br = returns[i]*backtest_df.loc[i, 'number_of_stocks']
    rp = br/per_stock_val*100
    backtest_returns.append(round(br, 2))
    return_percentage.append(round(rp, 2))
backtest_df['return'] = backtest_returns
backtest_df['return_percentage'] = return_percentage

backtest_df

Источник