Pythonによる米国株テクニカル分析と検証:¶

-どのテクニカル指標が良い結果を出すのか過去データを検証しよう¶

Pythonによる米国株テクニカル分析と検証: -どのテクニカル指標が良い結果を出すのか過去データを検証しよう
Satoshi (著)  形式: Kindle版
ASIN: B092LX6D25
発売日: 2021/4/14

本のソースコード: https://vrquest.org/index.php/technical-indicator-and-anomaly

Google Colab: https://colab.research.google.com/introp.ipython

筆者のTwitter: https://twitter.com/beavis28

筆者の Youtube: https://www.youtube.com/channel/UChXU9b0QMF24hw1Au-VAS0g

In [1]:
is_colab = 'google.colab' in str(get_ipython())   # for Google Colab

Alpha Vantage データを取り込む¶

データを取り込むライブラリとして Alpha Vantage を無料で利用する。

API Key を入手: https://www.alphavantage.co/suppoert/#api-key

Alpha Vantage API Documentation: https://www.alphavantage.co/documentation/

pypi Alpha Vantage API Documentation: https://pypi.org/project/alpha-vantage/

In [2]:
! pip -q install alpha_vantage
In [5]:
if not is_colab:
    ! pip -q install numpy matplotlib seaborn
In [7]:
if not is_colab:
    ! pip -q install scikit-learn
    ! pip -q install pandas
In [8]:
if not is_colab:
    ! pip -q install plotly
In [9]:
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split
from sklearn.tree import DecisionTreeClassifier
from datetime import date
from dateutil.relativedelta import relativedelta
from dateutil import parser
import numpy as np
import math
from alpha_vantage.techindicators import TechIndicators

import plotly.express as px
import plotly.figure_factory as ff
import plotly.graph_objects as go
import seaborn as sns
In [10]:
plt.style.use('fivethirtyeight')
plt.rcParams['figure.figsize'] = [24, 8]
API_KEY = 'Your API Key'     # YOU MUST CHANGE THIS.

Alpha Vantage から取得したデータから DataFrame を作る¶

  • Price_Up_From_PreviousDay ... 前日より値段が上がった/下がった (1/0)
  • Today_Price_Upその日の終値が上がった/下がった (1/0)
  • Tomorrow_Price_Up ... 翌日に値段が上がった/下がった (1/0)
  • Percentage ... 前日比での上下率
  • SomeDaysAfterClosingPrice ... 一定期間後の終値
In [11]:
symbol = "SNAP"
# symbol = "TWTR"
In [12]:
from alpha_vantage.timeseries import TimeSeries
from pprint import pprint
ti = TechIndicators(key=API_KEY, output_format='pandas')
ts = TimeSeries(key=API_KEY, output_format='pandas')
data, meta_data = ts.get_daily(symbol=symbol, outputsize='full')
data['Price_Up_From_PreviousDay'] = np.where(data['4. close'] > data['4. close'].shift(-1), 1, 0)
data['Today_Price_Up'] = np.where(data['4. close'] > data['1. open'], 1, 0)
data['Tomorrow_Price_Up'] = np.where(data['4. close'].shift(1) > data['4. close'], 1, 0)
data['Percentage'] = (data['4. close'] - data['4. close'].shift(-1)) / data['4. close'].shift(-1)
data['SomeDaysAfterClosingPrice'] = data['4. close'].shift(10) # X days later price
data
Out[12]:
1. open 2. high 3. low 4. close 5. volume Price_Up_From_PreviousDay Today_Price_Up Tomorrow_Price_Up Percentage SomeDaysAfterClosingPrice
date
2022-04-13 33.66 34.92 33.3401 34.68 17181342.0 1 1 0 0.033065 NaN
2022-04-12 34.72 35.95 33.1900 33.57 19538493.0 0 0 1 -0.024128 NaN
2022-04-11 34.65 35.75 34.0530 34.40 20890421.0 0 0 0 -0.035604 NaN
2022-04-08 35.88 36.83 35.4100 35.67 21718400.0 0 0 0 -0.016000 NaN
2022-04-07 36.39 37.30 34.5800 36.25 21490374.0 0 0 0 -0.006032 NaN
... ... ... ... ... ... ... ... ... ... ...
2017-03-08 22.03 23.43 21.3100 22.81 49834423.0 1 1 0 0.063899 21.82
2017-03-07 22.21 22.50 20.6400 21.44 71899652.0 0 0 1 -0.098023 20.38
2017-03-06 28.17 28.25 23.7700 23.77 72938848.0 0 0 0 -0.122554 19.93
2017-03-03 26.39 29.44 26.0600 27.09 148227379.0 1 1 0 0.106618 19.54
2017-03-02 24.00 26.05 23.5000 24.48 217109769.0 0 1 1 NaN 19.89

1290 rows × 10 columns

移動平均 (Moving Average)¶

買い時の指標のひとつ¶

ゴールデンクロス ... 短期間の移動平均が、長期間の移動平均を抜いた時

In [13]:
# Moving Average 5 and 20
dataSMA_short, meta_data_sma_short = ti.get_sma(symbol=symbol, time_period=5)
dataSMA_short.columns = ['SMA_short']
data = data.merge(dataSMA_short, left_index=True, right_index=True)

dataSMA_long, meta_data_sma_long = ti.get_sma(symbol=symbol, time_period=20)
dataSMA_long.columns = ['SMA_long']
data = data.merge(dataSMA_long, left_index=True, right_index=True)

data['SMATrend'] = np.where(data['SMA_short'] > data['SMA_long'], 1, 0) # Up trend is 1,, Down Trend is 0
data['GoldenCrossHappened'] = np.where(data['SMATrend'] > data['SMATrend'].shift(-1), 1, 0)
#pd.set_option('display.max_rows', None)

# theory 1
# we should see price up after seeing golden cross between SMA 5 and SMA20.
# this is giving as some days later price as example
turningPointChange = 0
priceActuallyUp = 0
for index, row in data.iterrows():
    if row['GoldenCrossHappened'] == 1:
      turningPointChange += 1
      if row['4. close'] < row['SomeDaysAfterClosingPrice']:
        priceActuallyUp += 1

print("Golden Cross Point:  " + str(turningPointChange))
print("Actual Price Up:  " + str(priceActuallyUp))
print("Percentage:  " + str((priceActuallyUp/turningPointChange)*100) + "%")

def interactive_plot(df, title):
  fig = px.line(title = title)
  
  # Loop through each stock (while ignoring time columns with index 0)
  for i in df.columns[:]:
    if i == "4. close" or i == "SMA_short" or i == "SMA_long":
      fig.add_scatter(x = df.index, y = df[i], name = i) # add a new Scatter trace

  fig.show()

interactive_plot(data, 'Prices')

Golden Cross Point:  36
Actual Price Up:  18
Percentage:  50.0%

ボリンジャーバンド¶

「価格の大半がボリンジャーバンド (幅) の中に収まる」が、バンドから離れた価格は10日後に戻るか？を検証する。

In [14]:
#BB
dataBB, meta_data_bb = ti.get_bbands(symbol=symbol)
data = data.merge(dataBB, left_index=True, right_index=True)
data['PriceBelowLowerBB'] = np.where(data['Real Lower Band'] > data['4. close'], 1, 0)
data['TouchDownOnLowerBBHappened'] = np.where(data['PriceBelowLowerBB'] > data['PriceBelowLowerBB'].shift(-1), 1, 0)
# theory 2
# we should see price up after reaching lower BB.
# this is giving as some days later price as example
turningPointChange = 0
priceActuallyUp = 0
for index, row in data.iterrows():
    if row['TouchDownOnLowerBBHappened'] == 1:
      turningPointChange += 1
      if row['4. close'] < row['SomeDaysAfterClosingPrice']:
        priceActuallyUp += 1

print("LowerBBTouched Point:  " + str(turningPointChange))
print("Actual Price Up:  " + str(priceActuallyUp))
print("Percentage:  " + str((priceActuallyUp/turningPointChange)*100) + "%")

def interactive_plot(df, title):
  fig = px.line(title = title)
  
  # Loop through each stock (while ignoring time columns with index 0)
  for i in df.columns[:]:
    if i == "4. close" or i == "Real Middle Band" or i == "Real Upper Band" or i == "Real Lower Band":
      fig.add_scatter(x = df.index, y = df[i], name = i) # add a new Scatter trace

  fig.show()

interactive_plot(data, 'Prices')

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
Input In [14], in <cell line: 2>()
      1 #BB
----> 2 dataBB, meta_data_bb = ti.get_bbands(symbol=symbol)
      3 data = data.merge(dataBB, left_index=True, right_index=True)
      4 data['PriceBelowLowerBB'] = np.where(data['Real Lower Band'] > data['4. close'], 1, 0)

File D:\sys\Anaconda3\envs\stock\lib\site-packages\alpha_vantage\alphavantage.py:218, in AlphaVantage._output_format.<locals>._format_wrapper(self, *args, **kwargs)
    216 @wraps(func)
    217 def _format_wrapper(self, *args, **kwargs):
--> 218     call_response, data_key, meta_data_key = func(
    219         self, *args, **kwargs)
    220     if 'json' in self.output_format.lower() or 'pandas' \
    221             in self.output_format.lower():
    222         if data_key is not None:

File D:\sys\Anaconda3\envs\stock\lib\site-packages\alpha_vantage\alphavantage.py:160, in AlphaVantage._call_api_on_func.<locals>._call_wrapper(self, *args, **kwargs)
    158 else:
    159     url = '{}{}'.format(url, apikey_parameter)
--> 160 return self._handle_api_call(url), data_key, meta_data_key

File D:\sys\Anaconda3\envs\stock\lib\site-packages\alpha_vantage\alphavantage.py:361, in AlphaVantage._handle_api_call(self, url)
    359     raise ValueError(json_response["Error Message"])
    360 elif "Information" in json_response and self.treat_info_as_error:
--> 361     raise ValueError(json_response["Information"])
    362 elif "Note" in json_response and self.treat_info_as_error:
    363     raise ValueError(json_response["Note"])

ValueError: Thank you for using Alpha Vantage! This is a premium endpoint and there are multiple ways to unlock premium endpoints: (1) become a holder of Alpha Vantage Coin (AVC), an Ethereum-based cryptocurrency that provides various utility & governance functions within the Alpha Vantage ecosystem (AVC mining guide: https://www.alphatournament.com/avc_mining_guide/) to unlock all premium endpoints, (2) subscribe to any of the premium plans at https://www.alphavantage.co/premium/ to instantly unlock all premium endpoints

RSI (相対力指数)¶

終値ベースで、直近の一定期間において上昇変動と下降変動のどちらの勢いが強いか。

In [15]:
# RSI
dataRSI, meta_data_rsi = ti.get_rsi(symbol=symbol)
data = data.merge(dataRSI, left_index=True, right_index=True)
data['RSITouched30'] = np.where(data['RSI'] < 30, 1, 0)
data['TouchDownRSI30Happened'] = np.where(data['RSITouched30'] > data['RSITouched30'].shift(-1), 1, 0)
# theory 3
# we should see price up after reaching RSI30.
# this is giving as some days later price as example
turningPointChange = 0
priceActuallyUp = 0
for index, row in data.iterrows():
    if row['TouchDownRSI30Happened'] == 1:
      turningPointChange += 1
      if row['4. close'] < row['SomeDaysAfterClosingPrice']:
        priceActuallyUp += 1

print("Price Touch RSI30 Point:  " + str(turningPointChange))
print("Actual Price Up:  " + str(priceActuallyUp))
print("Percentage:  " + str((priceActuallyUp/turningPointChange)*100) + "%")

Price Touch RSI30 Point:  12
Actual Price Up:  5
Percentage:  41.66666666666667%

MACD (Moving Average Convergence Divergence)¶

MACD のラインがシグナルを抜いた上で、ヒストグラムも上昇である10日後に価格は上昇するか？

In [16]:
#MACD
dataMACD, meta_data_macd = ti.get_macd(symbol=symbol, interval='daily')
data = data.merge(dataMACD, left_index=True, right_index=True)
data['MACDOverSignal'] = np.where(data['MACD'] > data['MACD_Signal'], 1, 0)
data['MACDHistUp'] = np.where(data['MACD_Hist'] > data['MACD_Hist'].shift(-2), 1, 0)
data['MACDOverSignalTrendHappened'] = np.where(data['MACDOverSignal'] > data['MACDOverSignal'].shift(-1), 1, 0)
data['MACDHistUpTrendHappened'] = np.where(data['MACDHistUp'] > data['MACDHistUp'].shift(-1), 1, 0)
data['MACDUpTrendHappened'] = np.where((data['MACDOverSignalTrendHappened'] == 1) & (data['MACDHistUpTrendHappened'] == 1), 1, 0)
# theory 4
# we should see price up after reaching MACD Trend is up.
# this is giving as some days later price as example
turningPointChange = 0
priceActuallyUp = 0
for index, row in data.iterrows():
    if row['MACDUpTrendHappened'] == 1:
      turningPointChange += 1
      if row['4. close'] < row['SomeDaysAfterClosingPrice']:
        priceActuallyUp += 1

print("MACD Uptrend Point:  " + str(turningPointChange))
print("Actual Price Up:  " + str(priceActuallyUp))
print("Percentage:  " + str((priceActuallyUp/turningPointChange)*100) + "%")

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
Input In [16], in <cell line: 2>()
      1 #MACD
----> 2 dataMACD, meta_data_macd = ti.get_macd(symbol=symbol, interval='daily')
      3 data = data.merge(dataMACD, left_index=True, right_index=True)
      4 data['MACDOverSignal'] = np.where(data['MACD'] > data['MACD_Signal'], 1, 0)

File D:\sys\Anaconda3\envs\stock\lib\site-packages\alpha_vantage\alphavantage.py:218, in AlphaVantage._output_format.<locals>._format_wrapper(self, *args, **kwargs)
    216 @wraps(func)
    217 def _format_wrapper(self, *args, **kwargs):
--> 218     call_response, data_key, meta_data_key = func(
    219         self, *args, **kwargs)
    220     if 'json' in self.output_format.lower() or 'pandas' \
    221             in self.output_format.lower():
    222         if data_key is not None:

File D:\sys\Anaconda3\envs\stock\lib\site-packages\alpha_vantage\alphavantage.py:160, in AlphaVantage._call_api_on_func.<locals>._call_wrapper(self, *args, **kwargs)
    158 else:
    159     url = '{}{}'.format(url, apikey_parameter)
--> 160 return self._handle_api_call(url), data_key, meta_data_key

File D:\sys\Anaconda3\envs\stock\lib\site-packages\alpha_vantage\alphavantage.py:361, in AlphaVantage._handle_api_call(self, url)
    359     raise ValueError(json_response["Error Message"])
    360 elif "Information" in json_response and self.treat_info_as_error:
--> 361     raise ValueError(json_response["Information"])
    362 elif "Note" in json_response and self.treat_info_as_error:
    363     raise ValueError(json_response["Note"])

ValueError: Thank you for using Alpha Vantage! This is a premium endpoint and there are multiple ways to unlock premium endpoints: (1) become a holder of Alpha Vantage Coin (AVC), an Ethereum-based cryptocurrency that provides various utility & governance functions within the Alpha Vantage ecosystem (AVC mining guide: https://www.alphatournament.com/avc_mining_guide/) to unlock all premium endpoints, (2) subscribe to any of the premium plans at https://www.alphavantage.co/premium/ to instantly unlock all premium endpoints

追加検証¶

曜日アノマリーを分析¶

「金曜日は売られやすく、月曜日は買われやすいか」を検証する

今回のデータでは「金曜日が最も買われやすい」という結論が出た。

In [17]:
#Calendar data
# which week of day get higher price than previous day
data = data.sort_index(ascending=1)
count = {'Monday': 0, 'Tuesday': 0, 'Wednesday': 0, 'Thursday': 0,'Friday': 0}
for index, row in data.iterrows():
  if index.day_name() == 'Monday' and int(row['Price_Up_From_PreviousDay']) == 1:
    count['Monday'] += 1
  elif index.day_name() == 'Tuesday' and int(row['Price_Up_From_PreviousDay']) == 1:
    count['Tuesday'] += 1
  elif index.day_name() == 'Wednesday' and int(row['Price_Up_From_PreviousDay']) == 1:
    count['Wednesday'] += 1
  elif index.day_name() == 'Thursday' and int(row['Price_Up_From_PreviousDay']) == 1:
    count['Thursday'] += 1  
  elif index.day_name() == 'Friday' and int(row['Price_Up_From_PreviousDay']) == 1:
    count['Friday'] += 1 

count
Out[17]:
{'Monday': 107,
 'Tuesday': 123,
 'Wednesday': 127,
 'Thursday': 131,
 'Friday': 136}

前日も翌日も連続して株価が上がる曜日を調べる。

今回のデータでは木曜日が最も株価が上がっている。

In [18]:
# Check which day of week get previous day and next day also got up
data = data.sort_index(ascending=1)
series_count = {'Monday': 0, 'Tuesday': 0, 'Wednesday': 0, 'Thursday': 0,'Friday': 0}
for index, row in data.iterrows():
  if index.day_name() == 'Monday' and int(row['Price_Up_From_PreviousDay']) == 1 and int(row['Tomorrow_Price_Up']) == 1:
    series_count['Monday'] += 1
  elif index.day_name() == 'Tuesday' and int(row['Price_Up_From_PreviousDay']) == 1 and int(row['Tomorrow_Price_Up']) == 1:
    series_count['Tuesday'] += 1
  elif index.day_name() == 'Wednesday' and int(row['Price_Up_From_PreviousDay']) == 1 and int(row['Tomorrow_Price_Up']) == 1:
    series_count['Wednesday'] += 1
  elif index.day_name() == 'Thursday' and int(row['Price_Up_From_PreviousDay']) == 1 and int(row['Tomorrow_Price_Up']) == 1:
    series_count['Thursday'] += 1  
  elif index.day_name() == 'Friday' and int(row['Price_Up_From_PreviousDay']) == 1 and int(row['Tomorrow_Price_Up']) == 1:
    series_count['Friday'] += 1  
series_count
Out[18]:
{'Monday': 43, 'Tuesday': 59, 'Wednesday': 67, 'Thursday': 69, 'Friday': 68}

前日比からの暴騰率を足し合わせた結果が、最も上昇しやすい曜日を調べる。

今回のデータでは、金曜日のスコアが一番よい。

In [19]:
# which week of day get higher average raise
percentageUp = {'Monday': 0, 'Tuesday': 0, 'Wednesday': 0, 'Thursday': 0,'Friday': 0}
total = {'Monday': 0, 'Tuesday': 0, 'Wednesday': 0, 'Thursday': 0,'Friday': 0}
for index, row in data.iterrows():
  if not math.isnan(row['Percentage']):
    if index.day_name() == 'Monday':
      percentageUp['Monday'] += row['Percentage']
      total['Monday'] += 1
    elif index.day_name() == 'Tuesday':
      percentageUp['Tuesday'] += row['Percentage']
      total['Tuesday'] += 1
    elif index.day_name() == 'Wednesday':
      percentageUp['Wednesday'] += row['Percentage']
      total['Wednesday'] += 1
    elif index.day_name() == 'Thursday':
      percentageUp['Thursday'] += row['Percentage']  
      total['Thursday'] += 1
    elif index.day_name() == 'Friday':
      percentageUp['Friday'] += row['Percentage'] 
      total['Friday'] += 1

percentageUp['Monday'] = percentageUp['Monday']/ total['Monday']*100
percentageUp['Tuesday'] = percentageUp['Tuesday']/ total['Tuesday']*100
percentageUp['Wednesday'] = percentageUp['Wednesday']/ total['Wednesday']*100
percentageUp['Thursday'] = percentageUp['Thursday']/ total['Thursday']*100
percentageUp['Friday'] = percentageUp['Friday']/ total['Friday']*100
percentageUp
Out[19]:
{'Monday': -0.37827719589172787,
 'Tuesday': 0.06393317369109724,
 'Wednesday': 0.23639491887426275,
 'Thursday': 0.08800158758023528,
 'Friday': 0.6133976315548666}

3 % 以上上昇した曜日をカウントする。

In [20]:
# which week of day get more than 3% raise

total = {'Monday': 0, 'Tuesday': 0, 'Wednesday': 0, 'Thursday': 0,'Friday': 0}
for index, row in data.iterrows():
  if not math.isnan(row['Percentage']) and row['Percentage'] > 0.03:
    if index.day_name() == 'Monday':

      total['Monday'] += 1
    elif index.day_name() == 'Tuesday':

      total['Tuesday'] += 1
    elif index.day_name() == 'Wednesday':

      total['Wednesday'] += 1
    elif index.day_name() == 'Thursday':
 
      total['Thursday'] += 1
    elif index.day_name() == 'Friday':

      total['Friday'] += 1

total
Out[20]:
{'Monday': 33, 'Tuesday': 41, 'Wednesday': 42, 'Thursday': 46, 'Friday': 50}
In [21]:
# historically how many days in a row they got price-up

data = data.sort_index(ascending=1)

days = 0
max_days = 0
percentage= 0
maxPercentage = 0
recorded_index = ""
recorded_index_row = ""
for index, row in data.iterrows():
  if not math.isnan(row['Percentage']):
    if row['Price_Up_From_PreviousDay'] == 1:
      days+= 1
      if days > max_days:
        max_days = days
        recorded_index_row = index

      percentage = row['Percentage']
      if percentage > maxPercentage:
        maxPercentage = percentage
        recorded_index = index
    else:
      #initialize
      days = 0   
      percentage = 0

print(recorded_index)
print("how much percentage they got up " + str(maxPercentage))
print(recorded_index_row)
print("how many days in row got up " + str(max_days))

2022-02-04 00:00:00
how much percentage they got up 0.5881632653061223
2020-06-23 00:00:00
how many days in row got up 8
In [22]:
# historically how many days in a row they got price-down

data = data.sort_index(ascending=1)

days = 0
max_days = 0
percentage= 0
maxPercentage = 0
recorded_index = ""
recorded_index_row = ""
for index, row in data.iterrows():
  if not math.isnan(row['Percentage']):
    if row['Price_Up_From_PreviousDay'] == 0:
      days+= 1
      if days > max_days:
        max_days = days
        recorded_index_row = index

      percentage = row['Percentage']*-1
      if percentage > maxPercentage:
        maxPercentage = percentage
        recorded_index = index
    else:

      #initialize
      days = 0   
      percentage = 0


print(recorded_index)
print("how much percentage they got down " + str(maxPercentage))
print(recorded_index_row)
print("how many days in row got down " + str(max_days))

2021-10-22 00:00:00
how much percentage they got down 0.2658767141525762
2022-01-27 00:00:00
how many days in row got down 11
In [ ]: