gavindiaz/vectorbt-visualization
1
0
Fork 0
mirror of https://github.com/awwesomeman/vectorbt-visualization.git synced 2026-07-27 18:57:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
tsunglin
2021-10-31 01:10:15 +08:00
parent 9b0bb4cdc4
commit 71f779f1ba
2 changed files with 530 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+111
View File
@@ -0,0 +1,111 @@
from plot import Plot
class My_Strategy(Plot):
def __init__(self,api):
super().__init__(api)
def KDJMA(self, time_period = 10,oversell = 20,overbuy = 80, ma1 = 5,ma2 = 20,trading_type='standard'):
#-------------------------------------------------
# calculate k, d , j
#-------------------------------------------------
data = self.kbars_df
ini_k = 50
ini_d = 50
k=[]
d=[]
rsv = (data['Close'].rolling(time_period).apply(lambda x:x[-1]) - data["Low"].rolling(time_period).min() ) / ( data["High"].rolling(time_period).max() - data["Low"].rolling(time_period).min() ) *100
rsv = rsv.dropna()
for _ in rsv:
ini_k = 2/3 * ini_k + 1/3 * _
k.append(ini_k)
for _ in k:
ini_d = 2/3 * ini_d + 1/3 * _
d.append(ini_d)
k = pd.Series(k,index = rsv.index)
d = pd.Series(d,index = rsv.index)
j = 3 * k - 2 * d
ma1 = data['Close'].rolling(ma1).mean()
ma2 = data['Close'].rolling(ma2).mean()
#-------------------------------------------------
# make strategy signals
#-------------------------------------------------
init_buy_sig = ( j>oversell ) & ( j.shift()<oversell ) & (ma1>ma2)
init_sell_sig = ( j<overbuy ) & ( j.shift()>overbuy ) & (ma1<ma2)
indicators = [('sub',{"K":k,"D":d,"J":j}),('main',{"ma1":ma1,"ma2":ma2})] # [('sub', {indicators1}), ('main'{indicators2})......]
self.init_buy_sig = init_buy_sig
self.init_sell_sig = init_sell_sig
self.indicators = indicators
self.trading_type = trading_type
def MA(self, ma1 = 5,ma2 = 20,trading_type='standard'):
#-------------------------------------------------
# calculate ma
#-------------------------------------------------
data = self.kbars_df
ma1 = data['Close'].rolling(ma1).mean()
ma2 = data['Close'].rolling(ma2).mean()
#-------------------------------------------------
# make strategy signals
#-------------------------------------------------
init_buy_sig = ( ma1>ma2 ) & ( ma1.shift()<ma2 )
init_sell_sig = ( ma1<ma2 ) & ( ma1.shift()>ma2 )
indicators = [('main',{"ma1":ma1,"ma2":ma2})] # [('sub', {indicators1}), ('main'{indicators2})......]
self.init_buy_sig = init_buy_sig
self.init_sell_sig = init_sell_sig
self.indicators = indicators
self.trading_type = trading_type
def KDJ(self, time_period = 10,oversell = 20,overbuy = 80,trading_type='standard'):
#-------------------------------------------------
# calculate k, d , j
#-------------------------------------------------
data = self.kbars_df
ini_k = 50
ini_d = 50
k=[]
d=[]
rsv = (data['Close'].rolling(time_period).apply(lambda x:x[-1]) - data["Low"].rolling(time_period).min() ) / ( data["High"].rolling(time_period).max() - data["Low"].rolling(time_period).min() ) *100
rsv = rsv.dropna()
for _ in rsv:
ini_k = 2/3 * ini_k + 1/3 * _
k.append(ini_k)
for _ in k:
ini_d = 2/3 * ini_d + 1/3 * _
d.append(ini_d)
k = pd.Series(k,index = rsv.index)
d = pd.Series(d,index = rsv.index)
j = 3 * k - 2 * d
#-------------------------------------------------
# make strategy signals
#-------------------------------------------------
init_buy_sig = ( j>oversell ) & ( j.shift()<oversell )
init_sell_sig = ( j<overbuy ) & ( j.shift()>overbuy )
# 模擬真實交易狀況，在訊號出現的下一個交易機會買入
init_buy_sig = init_buy_sig.shift().dropna()
init_sell_sig = init_sell_sig.shift().dropna()
indicators = [('sub',{"K":k,"D":d,"J":j})] # [('sub', {indicators1}), ('main'{indicators2})......]
self.init_buy_sig = init_buy_sig
self.init_sell_sig = init_sell_sig
self.indicators = indicators
self.trading_type = trading_type
plot.py
+419
View File
@@ -0,0 +1,419 @@
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import plotly.graph_objects as go
import shioaji as sj
from tabulate import tabulate
import yfinance as yf
import warnings
warnings.simplefilter(action='ignore', category=FutureWarning)
warnings.filterwarnings("ignore")
api = sj.Shioaji()
api.login("id_num", "password")
api.activate_ca(
ca_path="C:\\Users\\tsunglin\\Desktop\\SinoPac.pfx",
ca_passwd="id_num",
person_id="id_num")
class Plot():
def __init__(self,api=api):
self.api = api
def input_stock(self,kbars_df,stock_id,freq='B'):
if freq =='M':
label = 'left'
else:
label = 'right'
data = {
"Open":kbars_df["Open"].resample(freq,closed='right',label=label).first(),
"High":kbars_df["High"].resample(freq,closed='right',label=label).max(),
"Low":kbars_df["Low"].resample(freq,closed='right',label=label).min(),
"Close":kbars_df["Close"].resample(freq,closed='right',label=label).last(),
"Volume":kbars_df["Volume"].resample(freq,closed='right',label=label).sum()
}
kbars_df = pd.DataFrame(data,index=data['Open'].index).dropna()
self.freq = freq
self.stock_id = stock_id
self.kbars_df = kbars_df
def get_stock(self,stock_id,start,end,freq='B'):
'''
freq = Day -> 'B',
Minutes -> '10T',
Week -> 'W',
Month -> 'M', set label = 'left'
'''
if freq =='M':
label = 'left'
else:
label = 'right'
contract = self.api.Contracts.Stocks[stock_id]
kbars = self.api.kbars(contract, start=start, end=end)
kbars_df = pd.DataFrame({**kbars})
kbars_df.ts = pd.to_datetime(kbars_df.ts)
kbars_df = kbars_df.set_index('ts')
data = {
"Open":kbars_df["Open"].resample(freq,closed='right',label=label).first(),
"High":kbars_df["High"].resample(freq,closed='right',label=label).max(),
"Low":kbars_df["Low"].resample(freq,closed='right',label=label).min(),
"Close":kbars_df["Close"].resample(freq,closed='right',label=label).last(),
"Volume":kbars_df["Volume"].resample(freq,closed='right',label=label).sum()
}
kbars_df = pd.DataFrame(data,index=data['Open'].index).dropna()
kbars_df.index.name="Date"
self.freq = freq
self.stock_id = stock_id
self.kbars_df = kbars_df
def __strategy_type(self):
data = self.kbars_df
trading_type = self.trading_type
init_buy_sig = self.init_buy_sig
init_sell_sig = self.init_sell_sig
if trading_type == 'standard':
buy_sig=init_buy_sig.copy()
buy_sig[init_buy_sig]=1
buy_sig[init_sell_sig]=-1
strategy_sig = buy_sig.replace(0,np.nan).ffill()
strategy_buy = strategy_sig[strategy_sig.diff()==2]
strategy_sell = strategy_sig[strategy_sig.diff()==-2]
strategy_buy_mark = data[data.index.isin(strategy_buy.index)]
strategy_buy_mark['Signal'] = 1
strategy_sell_mark = data[data.index.isin(strategy_sell.index)]
strategy_sell_mark['Signal'] = 0
if trading_type =='customize':
pass
return strategy_sell_mark, strategy_buy_mark
def run(self,enter_comm=0.001425*0.28,exit_comm=0.001425*0.28+0.003,buy_at_kbar="Close", sell_at_kbar="Close"):
sell_mark, buy_mark = self.__strategy_type()
self.buy_mark = buy_mark
self.sell_mark = sell_mark
self.buy_at_kbar = buy_at_kbar
self.sell_at_kbar = sell_at_kbar
df = pd.concat([buy_mark, sell_mark],join='outer',sort=False).sort_index()
df['Holding periods'] = df.index.to_series().diff().shift(-1)
df['Profit'] = (df[sell_at_kbar].shift(-1)*(1-exit_comm) - df[buy_at_kbar]*(1+enter_comm)) / (df[buy_at_kbar]*(1+enter_comm))
df['Absolute value'] = df['Profit'].abs()
df['Gain / Loss'] = df['Profit']>0
df['Gain / Loss'] = df['Gain / Loss'].map({True:'$Gain$',False:'Loss'})
profit = df[df['Signal'] == 1][['Profit','Holding periods','Absolute value','Gain / Loss']]
profit = profit.dropna()
profit['Cumulative ret'] = (profit['Profit']+1).cumprod()
profit['Maximum draw drown'] = ((profit['Cumulative ret'] - profit['Cumulative ret'].cummax())/profit['Cumulative ret'].cummax())
average = (profit['Profit'].mean()) * 100
std = profit['Profit'].std()
win_rate = ((profit['Profit']>0).sum() / len(profit))*100
average_period = profit['Holding periods'].mean()
num_trade = len(profit)
data = {"Average return(%)":average,"Strategy std":std,"Number of trades":num_trade,"Winning rate(%)":win_rate,'Average holding periods':average_period}
strategy_info = pd.DataFrame(data,index=['Strategy info'])
profit[['Profit','Absolute value','Cumulative ret','Maximum draw drown']] = (profit[['Profit','Absolute value','Cumulative ret','Maximum draw drown']]).apply(lambda x:round(x,4))
strategy_info[['Average return(%)','Strategy std','Winning rate(%)']] = (strategy_info[['Average return(%)','Strategy std','Winning rate(%)']]).apply(lambda x:round(x,4))
print(tabulate(profit,headers=profit.columns, tablefmt='pretty'))
print(tabulate(strategy_info,headers=strategy_info.columns, tablefmt='pretty'))
self.trading_detail = df
self.profit_detail = profit
self.strategy_info = strategy_info
def plot(self):
df = self.kbars_df
freq = self.freq
stock_id = self.stock_id
indicator_set = self.indicators
buy_mark = self.buy_mark
sell_mark = self.sell_mark
profit_detail = self.profit_detail
buy_at_kbar = self.buy_at_kbar
sell_at_kbar = self.buy_at_kbar
#----------------------------------------------------------------------------------
# create plot
#----------------------------------------------------------------------------------
# candle stick, buy and sell signals (y6)
#----------------------------------------------------------------------------------
fig = go.Figure()
fig.add_trace(
go.Candlestick(x=df.index,
open=df['Open'],
high=df['High'],
low=df['Low'],
close=df ['Close'],
name="Stock Prices",
increasing={'line': {'color': '#DC3912'}},
decreasing={'line': {'color': '#222A2A'}},
yaxis='y6' ))
fig.add_trace(
go.Scatter(
x=buy_mark.index,
y=buy_mark[buy_at_kbar],
name="Buy Signal",
marker=dict(color="#0DF9FF", size=15),
mode="markers",
marker_symbol="triangle-up",
yaxis='y6'))
fig.add_trace(
go.Scatter(
x=sell_mark.index,
y=sell_mark[sell_at_kbar],
name="Sell Signal",
marker=dict(color="#778AAE", size=15),
mode="markers",
marker_symbol="triangle-down",
yaxis='y6'))
#----------------------------------------------------------------------------------
# technical indicator (y5 or y6)
#----------------------------------------------------------------------------------
if indicator_set:
for indicators in indicator_set:
if indicators[0]=="main":
for indicator_name, indicator in indicators[1].items():
fig.add_trace(
go.Scatter(
x=indicator.index,
y=indicator,
name=indicator_name,
yaxis='y6'))
if indicators[0]=="sub":
for indicator_name, indicator in indicators[1].items():
fig.add_trace(
go.Scatter(
x=indicator.index,
y=indicator,
name=indicator_name,
yaxis='y5'))
#----------------------------------------------------------------------------------
# volume (y4)
#----------------------------------------------------------------------------------
fig.add_trace(
go.Bar(
x=df.index,
y=df['Volume'],
yaxis='y4',
name='Volume',
marker_color='#795548'))
#----------------------------------------------------------------------------------
# individual profit plot (y3)
#----------------------------------------------------------------------------------
fig.add_trace(
go.Scatter(
x=profit_detail.index,
y=profit_detail.Profit*100,
marker_size = (profit_detail["Absolute value"])*100,
marker_color = profit_detail["Gain / Loss"].map({'$Gain$':'#D62728',"Loss":'#2CA02C'}),
marker_symbol='circle',
mode='markers',
yaxis='y3',
name = 'Gain/Loss for each trade(%)'))
#----------------------------------------------------------------------------------
# cumulative profit plot (y2)
#----------------------------------------------------------------------------------
fig.add_trace(
go.Scatter(
x=profit_detail.index,
y=profit_detail['Cumulative ret']*100,
name='Cumulative return(%)',
line_color='#3366CC',
yaxis='y2'))
#----------------------------------------------------------------------------------
# maximum drawdown (y)
#----------------------------------------------------------------------------------
fig.add_trace(
go.Scatter(
x=profit_detail.index,
y=profit_detail['Maximum draw drown']*100,
name='Maximum draw down(%)',
line_color='#DC3912',
yaxis='y'))
#----------------------------------------------------------------------------------
# setup layout
#----------------------------------------------------------------------------------
# y axis
#----------------------------------------------------------------------------------
fig.update_layout(
xaxis=dict(
autorange=True,
rangeslider=dict(
autorange=True,
),
type="date"
),
yaxis=dict(
anchor="x",
autorange=True,
domain=[0, 0.1],
linecolor="#DC3912",
mirror=True,
showline=True,
side="right",
tickfont={"color": "#DC3912"},
tickmode="auto",
ticks="",
titlefont={"color": "#DC3912"},
type="linear",
zeroline=False,
),
yaxis2=dict(
anchor="x",
autorange=True,
domain=[0.11, 0.2],
linecolor="#3366CC",
mirror=True,
showline=True,
side="right",
tickfont={"color": "#3366CC"},
tickmode="auto",
ticks="",
titlefont={"color": "#3366CC"},
type="linear",
zeroline=False,
),
yaxis3=dict(
anchor="x",
autorange=True,
domain=[0.21, 0.35],
linecolor="#222A2A",
mirror=True,
showline=True,
side="right",
tickfont={"color": "#222A2A"},
tickmode="auto",
ticks="",
titlefont={"color": "#222A2A"},
type="linear",
zeroline=True,
),
yaxis4=dict(
anchor="x",
autorange=True,
domain=[0.36, 0.45],
linecolor="#795548",
mirror=True,
showline=True,
side="right",
tickfont={"color": "#795548"},
tickmode="auto",
ticks="",
titlefont={"color": "#795548"},
type="linear",
zeroline=False,
),
yaxis5=dict(
anchor="x",
autorange=True,
domain=[0.46, 0.55],
linecolor="#673ab7",
mirror=True,
showline=True,
side="right",
tickfont={"color": "#673ab7"},
tickmode="auto",
ticks="",
titlefont={"color": "#673ab7"},
type="linear",
zeroline=False,
),
yaxis6=dict(
anchor="x",
autorange=True,
domain=[0.56, 1],
linecolor="#E91E63",
mirror=True,
showline=True,
side="right",
tickfont={"color": "#E91E63"},
tickmode="auto",
ticks="",
titlefont={"color": "#E91E63"},
type="linear",
zeroline=False,
))
#----------------------------------------------------------------------------------
# final settings
#----------------------------------------------------------------------------------
fig.update_traces(decreasing_line_width=2,increasing_line_width=2,whiskerwidth=0.5, selector=dict(type='candlestick'))
if freq=='B':
fig.update_xaxes(rangebreaks=[
dict(bounds=['sat', 'mon'])])
if freq[-1]=='T':
fig.update_xaxes(rangebreaks=[dict(bounds=[14, 9], pattern="hour"),
dict(bounds=['sat', 'mon'])])
fig.update_layout(
title = f'STOCK ID : {stock_id}, Data frequency:{freq}',
dragmode="zoom",
hovermode="x",
legend=dict(orientation="h",
yanchor="top",
xanchor="center",
y=-0.3,
x=0.5,
traceorder="reversed"),
height=1000,
width=1000,
template="plotly_white",
margin=dict(
l = 70,
r = 70,
t=100,
b=50
),
)
fig.show()