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
is_colab = 'google.colab' in str(get_ipython()) # for Google Colab
データを取り込むライブラリとして 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/
! pip -q install alpha_vantage
if not is_colab:
! pip -q install numpy matplotlib seaborn
if not is_colab:
! pip -q install scikit-learn
! pip -q install pandas
if not is_colab:
! pip -q install plotly
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
plt.style.use('fivethirtyeight')
plt.rcParams['figure.figsize'] = [24, 8]
API_KEY = 'Your API Key' # YOU MUST CHANGE THIS.
symbol = "SNAP"
# symbol = "TWTR"
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
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 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日後に戻るか?を検証する。
#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
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 のラインがシグナルを抜いた上で、ヒストグラムも上昇である10日後に価格は上昇するか?
#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
#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
{'Monday': 107, 'Tuesday': 123, 'Wednesday': 127, 'Thursday': 131, 'Friday': 136}
前日も翌日も連続して株価が上がる曜日を調べる。
今回のデータでは木曜日が最も株価が上がっている。
# 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
{'Monday': 43, 'Tuesday': 59, 'Wednesday': 67, 'Thursday': 69, 'Friday': 68}
前日比からの暴騰率を足し合わせた結果が、最も上昇しやすい曜日を調べる。
今回のデータでは、金曜日のスコアが一番よい。
# 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
{'Monday': -0.37827719589172787, 'Tuesday': 0.06393317369109724, 'Wednesday': 0.23639491887426275, 'Thursday': 0.08800158758023528, 'Friday': 0.6133976315548666}
3 % 以上上昇した曜日をカウントする。
# 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
{'Monday': 33, 'Tuesday': 41, 'Wednesday': 42, 'Thursday': 46, 'Friday': 50}
# 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
# 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