# 一、股池為0050成分股。
# 二、yoy為月營收年成長，對應日期為下個月10號的最近交易日。
# 三、近Y月至少發生一次 ，yoy > 0 且 上期yoy < 0
# 四、選擇滿足條件YOY較大的X檔股票
# 五、止損為Z% ,各個股票張數金額不超過A萬
import numpy as np
import pandas as pd
import kgisuperpy as kgi
person_id = 'A123456789'
person_pwd = '0000'
#simulation可輸入參數True：連線至測試環境；False：連線至正式環境
simulation = True
api = kgi.login(person_id ,person_pwd ,simulation)
msmp = api.MSMP
msmp_to_dic = api.MSMP.msmp_to_dic
x=8
y=3
z=15
a=100
BM='0050'
table=api.Data.get('ETF持股清單','0050')
ls = table[table.標的類型=='股票'].股票代號.tolist()
dic={}
for symbol in ls:
table1=api.Data.get('日K/技術指標/價量資料-個股多期',symbol)
df1=table1.set_index('日期')[['開盤價','收盤價','最高價','最低價','成交量']]
df1.columns=['Open','Close','High','Low','Volume']
table2 = api.Data.get('還原日K(個股、ETF、大盤)-單股單期',symbol)
df2 = table2.set_index('日期')[['開盤價','收盤價','最高價','最低價']]
df2.columns=['adjopen','adjclose','adjhigh','adjlow']
dic[symbol]= df1.sort_index().join(df2, how='inner')
def convert_to_datetime(date_str):
year, month = map(int, date_str.split('-M'))
day = 10
if month == 12:
year += 1
month = 1
else:
month += 1
return pd.Timestamp(year, month, day)
_yoy={}
for symbol in ls:
table = api.Data.get('月營收/月增率/年增率-個股多期',symbol)
df= table.set_index('年月')['單月合併營收年成長(%)']
_yoy[symbol]=df
yoy = pd.DataFrame(_yoy)
yoy.index = [convert_to_datetime(idx) for idx in yoy.index]
benchmark= api.Data.get('還原日K(個股、ETF、大盤)-單股單期' ,BM).set_index('日期')['收盤價']
index_ls = yoy.index
loaded_indices = benchmark.index
insert_indices = np.searchsorted(loaded_indices, index_ls)
yoy.index = [ loaded_indices[insert_indices[i]] if loaded_indices.min() <= index_ls[i] <= loaded_indices.max()
else index_ls[i] for i in range(len(index_ls))]
con = (yoy>0) & (yoy.shift(1)<0)
pos = con.rolling(y).sum()>=1
pos = pos*yoy
map_yoy = pos.where(pos > 0).rank(axis=1, ascending=False, method='first') <= x
dicX = {}
for stockid in dic.keys():
df = dic[stockid].copy()
df['sig'] = df.index.map(map_yoy[stockid])
df['qty'] = np.floor(a*10000/(df.Open*1000))*1000
if df.sig.any()!=True:
continue
dicX[stockid] = df.reset_index()
rp = api.backtest(data=dicX ,
date='日期',
sig_value='adjclose',
open_value='adjopen', open_shift=True,
close_value='adjopen', close_shift=True,
qty='qty',
longsig='sig',
loss= z,
benchmark=benchmark,
)
