代码拉取完成,页面将自动刷新
同步操作将从 lightning-trader/stock_robot 强制同步,此操作会覆盖自 Fork 仓库以来所做的任何修改,且无法恢复!!!
确定后同步将在后台操作,完成时将刷新页面,请耐心等待。
import pandas as pd
import cvt_helper as cvt
from stock_holder import StockHolder
from stock_info import StockSceneInfo
from stock_info import StockPredictInfo
import const as cst
from data_center import DataCenter
from stable_baselines3 import SAC
import const as cst
class StockPortfolioTest():
def __init__(self,model_path,is_safe):
self.current_step = cst.HISTORY_DATA_COUNT
self.is_safe = is_safe
self.model = SAC.load(model_path)
self.data_center = DataCenter()
def run(self,start_date,zs_type):
#self.protfolie_stock = self.data_center.query_concern_stock(start_date,zs_type)
#中证500 10大成分股
self.protfolie_stock = ["sh.600522","sh.601615","sh.600256","sh.600157","sz.002407","sz.000009","sz.002180","sz.002384"]
#沪深300 10大成分股
#self.protfolie_stock = ["sh.600519","sh.601318","sh.600036","sh.601012","sz.000858","sz.002594","sz.000333","sh.600900","sh.601166"]
#上证 10大成分股
#self.protfolie_stock = ["sh.600519","sh.601398","sh.601288","sh.601857","sh.600036","sh.601988","sh.601628","sh.600900","sh.601088","sh.601012"]
#方差最小10大股份
#self.protfolie_stock = ["sh.601880","sz.002936","sh.600022","sh.600567","sh.600871","sh.601860","sh.601005","sz.002958","sz.002948","sh.601992"]
if self.protfolie_stock is None :
print('run error : protfolie_stock is none')
return
total_delta_monty = 0
win_count = 0
max_win_money = 0
max_lose_money = 0
for it in self.protfolie_stock :
protfolie_stock = [it]
self.current_step = 0
self.stock_holder = StockHolder(self.is_safe,True)
self.all_stock = self.data_center.get_protfolie_info(protfolie_stock,start_date)
while not self._is_done():
self._step()
delta_monty = self.stock_holder.get_fortune()-cst.INITIAL_CAPITAL
if delta_monty > 0:
win_count += 1
if delta_monty > max_win_money:
max_win_money = delta_monty
else:
if delta_monty < -max_lose_money:
max_lose_money = -delta_monty
total_delta_monty+=delta_monty
total_num = len(self.protfolie_stock)
return win_count/total_num,max_win_money/cst.INITIAL_CAPITAL,max_lose_money/cst.INITIAL_CAPITAL,total_delta_monty/(total_num*cst.INITIAL_CAPITAL)
def _step(self):
for key,value in self.all_stock.items () :
frame_stock_data,last_info = self._get_frame_data(value)
self.stock_holder.update(key,last_info)
obs = cvt.get_obs(frame_stock_data)
action, _ = self.model.predict(obs,deterministic=True)
buy_price,sell_price = cvt.parse_action(last_info.Standard,action)
if key in self.stock_holder.stock_info :
self.stock_holder.sell(key,sell_price)
if last_info is not None:
self.stock_holder.buy(key,last_info,buy_price)
self.current_step += 1
def _print(self,stock_code):
print(f'{stock_code} Money : {self.stock_holder.get_fortune()},Safe Money : {self.stock_holder.safe_money}')
pass
def _is_done(self):
max_step = 0
for value in self.all_stock.values () :
if max_step < len(value)-1:
max_step = len(value)-1
return self.current_step>=max_step
def _get_frame_data(self,current_stock_data):
scene_info = []
for i in range(cst.HISTORY_DATA_COUNT):
scene : StockSceneInfo = current_stock_data[self.current_step-cst.HISTORY_DATA_COUNT+i]
scene_info.append(scene)
last_info:StockSceneInfo = current_stock_data[self.current_step]
return scene_info,last_info
if __name__ == "__main__":
model_name = "best_model_all_128c_2000_4"
test = StockPortfolioTest("./model/"+model_name,False)
result = []
#result.append(test.run("2022-01-01","sz50"))
result.append(test.run("2022-01-01","zz500"))
#result.append(test.run("2022-01-01","hs300"))
'''
result.append(test.run("2021-08-01","sz50"))
result.append(test.run("2021-08-01","zz500"))
result.append(test.run("2021-08-01","hs300"))
result.append(test.run("2020-08-01","sz50"))
result.append(test.run("2020-08-01","zz500"))
result.append(test.run("2020-08-01","hs300"))
'''
result = pd.DataFrame(result)
print(f"{model_name} test result")
print(result)
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。