supertrend indicator plot python
5 min readAug 22, 2021
ตัวอย่างใช้โบรค FTX เชื่อมด้วย library CCXT
วิธีดึงข้อมูล Historical Data ตลาดไทย ด้วย Yahoo Finance
เรียกใช้ library
import pandas as pdimport numpy as npimport matplotlib.pyplot as pltimport requestsfrom math import floorfrom termcolor import colored as climport ccxtplt.style.use('fivethirtyeight')plt.rcParams['figure.figsize'] = (20,10)
สร้าง functions เรียกราคาจาก FTX
ftx = ccxt.ftx({‘api_key’: ‘************************’,
‘secret’: ‘*************************’})
symbol = 'SOL/USD'
tf = '1h'
n = 500def get_historical_data(n,timeframe,product):
ohlc = ftx.fetch_ohlcv(product,timeframe=timeframe,limit=n)
ohlc_df = pd.DataFrame(ohlc, columns =['datetime', 'open','high','low','close','volume'])
ohlc_df['datetime'] = pd.to_datetime(ohlc_df['datetime'], unit='ms')
return ohlc_df.set_index('datetime')
สร้าง function คำนวณ Supertrend Indicator จาก historical data
def get_supertrend(high, low, close, lookback, multiplier):
# ATR
tr1 = pd.DataFrame(high - low)
tr2 = pd.DataFrame(abs(high - close.shift(1)))
tr3 = pd.DataFrame(abs(low - close.shift(1)))
frames = [tr1, tr2, tr3]
tr = pd.concat(frames, axis = 1, join = 'inner').max(axis = 1)
atr = tr.ewm(lookback).mean()
# H/L AVG AND BASIC UPPER & LOWER BAND
hl_avg = (high + low) / 2
upper_band = (hl_avg + multiplier * atr).dropna()
lower_band = (hl_avg - multiplier * atr).dropna()
# FINAL UPPER BAND
final_bands = pd.DataFrame(columns = ['upper', 'lower'])
final_bands.iloc[:,0] = [x for x in upper_band - upper_band]
final_bands.iloc[:,1] = final_bands.iloc[:,0]
for i in range(len(final_bands)):
if i == 0:
final_bands.iloc[i,0] = 0
else:
if (upper_band[i] < final_bands.iloc[i-1,0]) | (close[i-1] > final_bands.iloc[i-1,0]):
final_bands.iloc[i,0] = upper_band[i]
else:
final_bands.iloc[i,0] = final_bands.iloc[i-1,0]
# FINAL LOWER BAND
for i in range(len(final_bands)):
if i == 0:
final_bands.iloc[i, 1] = 0
else:
if (lower_band[i] > final_bands.iloc[i-1,1]) | (close[i-1] < final_bands.iloc[i-1,1]):
final_bands.iloc[i,1] = lower_band[i]
else:
final_bands.iloc[i,1] = final_bands.iloc[i-1,1]
# SUPERTREND
supertrend = pd.DataFrame(columns = [f'supertrend_{lookback}'])
supertrend.iloc[:,0] = [x for x in final_bands['upper'] - final_bands['upper']]
for i in range(len(supertrend)):
if i == 0:
supertrend.iloc[i, 0] = 0
elif supertrend.iloc[i-1, 0] == final_bands.iloc[i-1, 0] and close[i] < final_bands.iloc[i, 0]:
supertrend.iloc[i, 0] = final_bands.iloc[i, 0]
elif supertrend.iloc[i-1, 0] == final_bands.iloc[i-1, 0] and close[i] > final_bands.iloc[i, 0]:
supertrend.iloc[i, 0] = final_bands.iloc[i, 1]
elif supertrend.iloc[i-1, 0] == final_bands.iloc[i-1, 1] and close[i] > final_bands.iloc[i, 1]:
supertrend.iloc[i, 0] = final_bands.iloc[i, 1]
elif supertrend.iloc[i-1, 0] == final_bands.iloc[i-1, 1] and close[i] < final_bands.iloc[i, 1]:
supertrend.iloc[i, 0] = final_bands.iloc[i, 0]
supertrend = supertrend.set_index(upper_band.index)
supertrend = supertrend.dropna()[1:]
# ST UPTREND/DOWNTREND
upt = []
dt = []
close = close.iloc[len(close) - len(supertrend):]for i in range(len(supertrend)):
if close[i] > supertrend.iloc[i, 0]:
upt.append(supertrend.iloc[i, 0])
dt.append(np.nan)
elif close[i] < supertrend.iloc[i, 0]:
upt.append(np.nan)
dt.append(supertrend.iloc[i, 0])
else:
upt.append(np.nan)
dt.append(np.nan)
st, upt, dt = pd.Series(supertrend.iloc[:, 0]), pd.Series(upt), pd.Series(dt)
upt.index, dt.index = supertrend.index, supertrend.index
return st, upt, dtdef implement_st_strategy(prices, st):
buy_price = []
sell_price = []
st_signal = []
signal = 0
for i in range(len(st)):
if st[i-1] > prices[i-1] and st[i] < prices[i]:
if signal != 1:
buy_price.append(prices[i])
sell_price.append(np.nan)
signal = 1
st_signal.append(signal)
else:
buy_price.append(np.nan)
sell_price.append(np.nan)
st_signal.append(0)
elif st[i-1] < prices[i-1] and st[i] > prices[i]:
if signal != -1:
buy_price.append(np.nan)
sell_price.append(prices[i])
signal = -1
st_signal.append(signal)
else:
buy_price.append(np.nan)
sell_price.append(np.nan)
st_signal.append(0)
else:
buy_price.append(np.nan)
sell_price.append(np.nan)
st_signal.append(0)
return buy_price, sell_price, st_signal
เรียก plot
Full Code
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import requests
from math import floor
from termcolor import colored as cl
import ccxtplt.style.use('fivethirtyeight')
plt.rcParams['figure.figsize'] = (20,10)ftx = ccxt.ftx({
'api_key': '',
'secret': ''})def get_historical_data(n,timeframe,product):
ohlc = ftx.fetch_ohlcv(product,timeframe=timeframe,limit=n)
ohlc_df = pd.DataFrame(ohlc, columns =['datetime', 'open','high','low','close','volume'])
ohlc_df['datetime'] = pd.to_datetime(ohlc_df['datetime'], unit='ms')
return ohlc_df.set_index('datetime')def get_supertrend(high, low, close, lookback, multiplier):
# ATR
tr1 = pd.DataFrame(high - low)
tr2 = pd.DataFrame(abs(high - close.shift(1)))
tr3 = pd.DataFrame(abs(low - close.shift(1)))
frames = [tr1, tr2, tr3]
tr = pd.concat(frames, axis = 1, join = 'inner').max(axis = 1)
atr = tr.ewm(lookback).mean()
# H/L AVG AND BASIC UPPER & LOWER BAND
hl_avg = (high + low) / 2
upper_band = (hl_avg + multiplier * atr).dropna()
lower_band = (hl_avg - multiplier * atr).dropna()
# FINAL UPPER BAND
final_bands = pd.DataFrame(columns = ['upper', 'lower'])
final_bands.iloc[:,0] = [x for x in upper_band - upper_band]
final_bands.iloc[:,1] = final_bands.iloc[:,0]
for i in range(len(final_bands)):
if i == 0:
final_bands.iloc[i,0] = 0
else:
if (upper_band[i] < final_bands.iloc[i-1,0]) | (close[i-1] > final_bands.iloc[i-1,0]):
final_bands.iloc[i,0] = upper_band[i]
else:
final_bands.iloc[i,0] = final_bands.iloc[i-1,0]
# FINAL LOWER BAND
for i in range(len(final_bands)):
if i == 0:
final_bands.iloc[i, 1] = 0
else:
if (lower_band[i] > final_bands.iloc[i-1,1]) | (close[i-1] < final_bands.iloc[i-1,1]):
final_bands.iloc[i,1] = lower_band[i]
else:
final_bands.iloc[i,1] = final_bands.iloc[i-1,1]
# SUPERTREND
supertrend = pd.DataFrame(columns = [f'supertrend_{lookback}'])
supertrend.iloc[:,0] = [x for x in final_bands['upper'] - final_bands['upper']]
for i in range(len(supertrend)):
if i == 0:
supertrend.iloc[i, 0] = 0
elif supertrend.iloc[i-1, 0] == final_bands.iloc[i-1, 0] and close[i] < final_bands.iloc[i, 0]:
supertrend.iloc[i, 0] = final_bands.iloc[i, 0]
elif supertrend.iloc[i-1, 0] == final_bands.iloc[i-1, 0] and close[i] > final_bands.iloc[i, 0]:
supertrend.iloc[i, 0] = final_bands.iloc[i, 1]
elif supertrend.iloc[i-1, 0] == final_bands.iloc[i-1, 1] and close[i] > final_bands.iloc[i, 1]:
supertrend.iloc[i, 0] = final_bands.iloc[i, 1]
elif supertrend.iloc[i-1, 0] == final_bands.iloc[i-1, 1] and close[i] < final_bands.iloc[i, 1]:
supertrend.iloc[i, 0] = final_bands.iloc[i, 0]
supertrend = supertrend.set_index(upper_band.index)
supertrend = supertrend.dropna()[1:]
# ST UPTREND/DOWNTREND
upt = []
dt = []
close = close.iloc[len(close) - len(supertrend):]for i in range(len(supertrend)):
if close[i] > supertrend.iloc[i, 0]:
upt.append(supertrend.iloc[i, 0])
dt.append(np.nan)
elif close[i] < supertrend.iloc[i, 0]:
upt.append(np.nan)
dt.append(supertrend.iloc[i, 0])
else:
upt.append(np.nan)
dt.append(np.nan)
st, upt, dt = pd.Series(supertrend.iloc[:, 0]), pd.Series(upt), pd.Series(dt)
upt.index, dt.index = supertrend.index, supertrend.index
return st, upt, dtdef implement_st_strategy(prices, st):
buy_price = []
sell_price = []
st_signal = []
signal = 0
for i in range(len(st)):
if st[i-1] > prices[i-1] and st[i] < prices[i]:
if signal != 1:
buy_price.append(prices[i])
sell_price.append(np.nan)
signal = 1
st_signal.append(signal)
else:
buy_price.append(np.nan)
sell_price.append(np.nan)
st_signal.append(0)
elif st[i-1] < prices[i-1] and st[i] > prices[i]:
if signal != -1:
buy_price.append(np.nan)
sell_price.append(prices[i])
signal = -1
st_signal.append(signal)
else:
buy_price.append(np.nan)
sell_price.append(np.nan)
st_signal.append(0)
else:
buy_price.append(np.nan)
sell_price.append(np.nan)
st_signal.append(0)
return buy_price, sell_price, st_signalsymbol = 'SOL/USD'
tf = '1h'
n = 500OHLC = get_historical_data(n,tf,symbol)
OHLCOHLC['st'], OHLC['s_upt'], OHLC['st_dt'] = get_supertrend(OHLC['high'], OHLC['low'], OHLC['close'], 10, 3)
OHLC = OHLC[1:]
# print(BTC.head())buy_price, sell_price, st_signal = implement_st_strategy(OHLC['close'], OHLC['st'])plt.plot(OHLC['close'], linewidth = 2)
plt.plot(OHLC['st'], color = 'green', linewidth = 2, label = 'ST UPTREND')
plt.plot(OHLC['st_dt'], color = 'r', linewidth = 2, label = 'ST DOWNTREND')plt.plot(OHLC.index, buy_price, marker = '^', color = 'green', markersize = 12, linewidth = 0, label = 'BUY SIGNAL')
plt.plot(OHLC.index, sell_price, marker = 'v', color = 'r', markersize = 12, linewidth = 0, label = 'SELL SIGNAL')plt.title(f'{symbol} ST TRADING SIGNALS')
plt.legend(loc = 'upper left')
plt.show()