标签：plow py code6 df self huo dt def
代码
# -*- coding: utf-8 -*-
import os, math;
import openpyxl
import datetime, time;
import pandas as pd, numpy as np
from collections import namedtuple
_=(os, math, datetime, time, np)
import cnt
from pytdx2 import reader as rd
import cbond
rjfw = '880493' #软件服务指数
cn_dict_2jhy = {
'880301': '煤炭',
'880305': '电力',
'880310': '石油',
'880318': '钢铁',
'880324': '有色',
'880330': '化纤',
'880335': '化工',
'880344': '建材',
'880350': '造纸',
'880351': '矿物制品',
'880355': '日用化工',
'880360': '农林牧渔',
'880367': '纺织服饰',
'880372': '食品饮料',
'880380': '酿酒',
'880387': '家用电器',
'880390': '汽车类',
'880398': '医疗保健',
'880399': '家居用品',
'880400': '医药',
'880406': '商业连锁',
'880414': '商贸代理',
'880418': '传媒娱乐',
'880421': '广告包装',
'880422': '文教休闲',
'880423': '酒店餐饮',
'880424': '旅游',
'880430': '航空',
'880431': '船舶',
'880432': '运输设备',
'880437': '通用机械',
'880440': '工业机械',
'880446': '电气设备',
'880447': '工程机械',
'880448': '电器仪表',
'880452': '电信运营',
'880453': '公共交通',
'880454': '水务',
'880455': '供气供热',
'880456': '环境保护',
'880459': '运输服务',
'880464': '仓储物流',
'880465': '交通设施',
'880471': '银行',
'880472': '证券',
'880473': '保险',
'880474': '多元金融',
'880476': '建筑',
'880482': '房地产',
'880489': 'IT设备',
'880490': '通信设备',
'880491': '半导体',
'880492': '元器件',
'880493': '软件服务',
'880494': '互联网',
'880497': '综合类'}
# =============================================================================
# 通达信2级行业指数的: 指数代码和名称字典
# import cbond
# >>> self = cbond.BlockCfgList()
# >>> self._get_gp_hy_df(1)
# >>> cn_dict_2jhy = dict(zip(self.industry_df['code'], self.industry_df['name']))
# =============================================================================
def _get_mkt_id(code6):
return cbond._get_mkt_code(code6)[-2:]
def read_EOD_day(code6, type_=0):
'''读取通达信本地的日线数据文件
type_: 0表示证券品种, 1:表示交易所发布的指数型品种
'''
reader = rd.TdxDailyBarReader()
_dir = 'c:\\GTJA\\RichEZ\\newVer\\vipdoc'
_dic = {0:'sz', 1:'sh', 2:'bj'}
mkt_id = _get_mkt_id(code6)
if type(mkt_id)==int:
fname = _dir + '\\{0}\lday\\{0}{1}.day'.format(
_dic[mkt_id], code6)
elif type(mkt_id)==str:
fname = _dir + '\\{0}\lday\\{0}{1}.day'.format(
mkt_id, code6)
try:
df = reader.get_df_by_file(fname)
return df
except:
pass
return None
class ZQ:
'''
数据加载类:
- 只针对可交易的证券, 比如股票可转债基金等金融产品, 不包括指数
- 实例化就表示把该代码对应的证券数据存储在它的类的datalist属性里
小技巧: 将每一个指数的数据加载封装在类的实例化过程中, 简单明了.
'''
datalist = []
def __init__(self, code6):
self.code = code6
self.ohlcdf = None
code9 = cnt.code9(code6)
if code9 in cnt.cn_dict:
self.name = cnt.cn_dict[code9]
df = read_EOD_day(code6)
# df.index = df.index.format(date_format='%Y%m%d')
# df['roc1'] = df['close'].pct_change(1)
# df['roc30'] = df['close'].pct_change(30)
# self.ohlcdict = df.to_dict(orient='index')
if df is not None:
if len(df)>0:
self.ohlcdf = df
ZQ.datalist.append(self)
class Hyzs:
'''通达信行业指数类
用来加载880系列指数数据, 存储在它的datalist属性里
小技巧: 将每一个指数的数据加载封装在类的实例化过程中, 简单明了.
'''
datalist = []
def __init__(self, code6):
self.code, self.name = code6, cn_dict_2jhy[code6]
self.ohlcdf = None
df = read_EOD_day(code6)
# df.index = df.index.format(date_format='%Y%m%d')
# df['roc1'] = df['close'].pct_change(1)
# df['roc30'] = df['close'].pct_change(30)
# self.ohlcdict = df.to_dict(orient='index')
self.ohlcdf = df
if len(self.ohlcdf)>0:
Hyzs.datalist.append(self)
def doesHave(self, timestamp):
# return self.ohlcdict.__contains__(timestamp)
return timestamp in self.ohlcdf.index
def load_56_880hyidx(num=56, reset=1):
'''
类属性Hyzs.datalist被设计为存储56个行业指数的list,
该属性通常用于记录某一项需要保持不变的数据；
>>> load_56_880hyidx()
'''
codes = list(cn_dict_2jhy.keys())[:num]
if reset:
Hyzs.datalist.clear() #类属性要用类来操作, 不能用实例来操作
print('------>>>>>> Loading 通达信行业指数数据 ...')
for c in codes:
zs = Hyzs(c)
print(f'{zs.code}_{zs.name}', end=', ')
print() #消除上面的for循环里的print里面的不换行输出
#%% Account
class Bid:
'''申报价格, 出价'''
collection = []
Order = namedtuple('Order', 'sn dt time code name operation price volume')
sn = 0
time_ = '14:59:59'
volume = 1
def __init__(self, dt, code6, operation, price):
name = cn_dict_2jhy[code6]
self.order = Bid.Order(Bid.sn, dt, Bid.time_,
code6, name,
operation, price, Bid.volume)
Bid.collection.append(self)
Bid.sn += 1
class Account:
def __init__(self, write_excel, fname):
self.write_excel = write_excel
self.fname =fname
self.capital = 0.0 # 最大进货本金
self.cash = 0.0 # 现金余额
# 进出货备忘/进货单出货单, 存货备忘
self.order, self.holds = '', ''
#证券头寸字典/货柜 {品种对象:持有数量}, 最新存货和历史存货
self.holding, self.holdhist = {}, []
self.totmktcap = 0.0 #市值
self.totcap = 0.0 #总资产
if self.write_excel==True:
self.wb = openpyxl.Workbook()
self.sheet = self.wb.active
title = '日期 现金 总市值 总资产 进出单 货存'.split()
self.sheet.append(title)
def _buy(self, dt, huo, amount=None, vol=None):
# print('+', end=''); print(dt, ' buy ', huo.name)
self.holding[huo] = 1 #持仓字典里添加一个货,
# 持仓量为1个单位, 持仓成本: 收盘价格上浮一个手续费
price = float(huo.ohlcdf.loc[dt]['close'])
self.cash -= price * (1+0.0001 )
_ = Bid(dt, huo.code, 'B', price)
def _sell(self, dt, huo): #时间戳, 货
# print('-', end=''); print(dt, ' Sell ', huo.name)
if huo.doesHave(dt):
price = float(huo.ohlcdf.loc[dt]['close'])
self.cash += price * (1-0.0001)
_ = Bid(dt, huo.code, 'S', price)
else:
# dt = list(huo.ohlcdf.keys())[0] # ??
dt = huo.ohlcdf.index[0] # ??
self.cash += float(huo.ohlcdf.iloc[0]['close']) * (1-0.0001)
del self.holding[huo]
def _sell_all(self, dt):
'''清仓, 主循环结束后, 如果认为现金为王, 则会用到'''
self.order=''
for h in list(self.holding.keys()):
self._sell(dt, h)
self.order += f'S:{h.name}, '
self.holds=''
self.remark_huo(dt)
self.calc_totcap(dt)
self.recording(dt)
def write(self):
if self.write_excel==True:
self.wb.save('./data/随机漫步模拟交易_' + self.fname+'.xlsx')
def calc_totcap(self, dt):
self.totmktcap = 0
for h in list(self.holding.keys()):
self.totmktcap += float(h.ohlcdf.loc[dt]['close']) * self.holding[h]
self.totcap = self.totmktcap + self.cash
def sell_old__buy_new(self, dt, huo):
'''调仓/进出货, 并填写进出货的备忘录'''
order = ''
for h in list(self.holding.keys()):
if isinstance(huo, list):
# 多品种
if huo.count(h)==0:
self._sell(dt, h)
order += f'S:{h.name}, '
# elif isinstance(huo, Hyzs):
# # 单一品种
# pass
if isinstance(huo, list):
# 多品种货
for h in huo:
if not self.holding.__contains__(h):
self._buy(dt, h)
order += f'B:{h.name}, '
# elif isinstance(huo, Hyzs):
# # 单一品种货
# self._buy(dt, huo)
# order += f'B:{h.name}, '
self.order = order
def remark_huo(self, dt):
'''填写持仓/存货备忘录, 仅备忘货名或者货号, 不包含货数据'''
self.holds = 'H:'
for h in list(self.holding.keys()):
self.holds += f' {h.name}'
self.holdhist.append((dt, self.holds))
def recording(self, dt):
# if self.write_excel==True:
if True: # 应该总是要输出历史 总资产的
row = [dt,
round(self.cash, 3),
round(self.totmktcap, 3),
round(self.totcap, 3),
self.order, self.holds]
self.sheet.append(row)
def process(self, dt, huo=None, case=1):
'''处理头寸, 调仓操作:
依据当前的持仓品种, 与策略选出的目标头寸进行比较, 吐故纳新.
case: 有两种场景需要分别对待:
case==1: 需要调仓/有仓位操作的
case==0: 持股日, 无需调仓
'''
if case==1 and huo:
# 调仓: 有买卖
self.sell_old__buy_new(dt, huo)
self.remark_huo(dt)
self.calc_totcap(dt)
self.recording(dt)
if self.capital<abs(self.cash):
self.capital = abs(self.cash) #本金, 所需cash绝对值里的最大的值
elif case==0:
# 持股日: 持仓不动, 记录市值
self.remark_huo(dt) #备忘存货, 计算市值,
self.calc_totcap(dt)
self.order=''
self.recording(dt)
def calc_nav(self, begin=None, end=None, plot=0):
codes = [h.code for h in Hyzs.datalist]
names = [h.name for h in Hyzs.datalist]
df = pd.DataFrame()
for code, name in zip(codes, names):
ohlc_ = read_EOD_day(code)
df[name] = ohlc_['close']
df = df[begin:end]
df = df.apply(lambda x: x/x[0], axis=0)
if begin:
# beginD, endD = date8_to_date(begin), date8_to_date(end)
# dt = pd.DatetimeIndex(ohlcdict.keys())
# dt = dt[dt.get_loc(begin):dt.get_loc(end)]
A = [cell.value for cell in self.sheet['A']]
D = [cell.value for cell in self.sheet['D']]
cols = [A[0], D[0]]
data = [e for e in zip(A[1:], D[1:])]
df1 = pd.DataFrame(data, columns=cols)
df1['nav'] = (df1['总资产']/self.capital + 1)
df1 = df1.set_index('日期')
df1.index = pd.DatetimeIndex(df1.index)
df = df.join(df1.nav)
df = df.fillna(method='bfill').fillna(method='ffill')
if plot:
df.plot()
self.nav =df
def reset_Bid(self):
self.orderhist = pd.DataFrame([b.order for b in Bid.collection])
Bid.collection = []
Bid.sn = 0
def explore(self):
keys = [k for k in list(self.__dict__.keys())]
for k in keys:
v = self.__dict__[k]
type_str = str(type(v))[6:-1]
if '.' in type_str:
type_str = type_str.split('.')[-1]
try:
_ = len(v)
if isinstance(v, str):
print(f'{k:12s}', f'{type_str:10s}', v)
else:
print(f'{k:12s}', f'{type_str:10s}\n', v[:3])
except:
# 简单类型的数据比如浮点数, 不可以len()操作
# print(k, types[k], acnt.__getattribute__(k))
print(f'{k:12s}', f'{type_str:12s}', v)
pass
def is_trade_day(date_):
'''检查某天是否是休市日
date_: datetime类,
'''
if date_.isoweekday() > 5: # 排除周六和周日
return False
dates = date_.strftime('%Y%m%d')
zs = Hyzs.datalist[0]
if 1:
# for zs in Hyzs.datalist: # 只需要检查一个行业指数的节假日就可以了
if zs.doesHave(dates):
return True
else: # 节假日也是需要排除的
return False
kp_list = [9999999,
340,
336,
309,
291,
277,
248,
223,
163,
124,
99,
86,
81,
67,
63,
54,
29,
26,
23,
18,
5]
#%% 辅助函数
def date8_to_date(s):
'''日期字符串8位的, 转变成日期对象'''
yyyy, mm, dd = int(s[:4]), int(s[4:6]), int(s[-2:])
return datetime.datetime(yyyy, mm, dd)
to_date = date8_to_date
def gen_hold_idle(n=30):
'''随机生成n个自然日的'持股待涨'和'空仓等待'的时间节拍'''
nhold_days = int(np.random.uniform(1, n, 1))
nidle_days = n - nhold_days
print(nhold_days, nidle_days)
return (nhold_days, nidle_days)
def _get_random_target(n=2, industry=55):
"""基于通达信的56个行业指数, 随机选出n个行业作为标的头寸
返回值: 行业指数对象, 类型: list
>>> get_random_target()
"""
# npos = int(gs.np.random.uniform(1, 3, 1))
_list = [int(n) for n in np.random.uniform(0, 55, n)]
if len(Hyzs.datalist)<56:
print('行业指数数据可能没有加载完全')
tgt_list = [Hyzs.datalist[t] for t in _list]
print([t.name for t in tgt_list])
return tgt_list
def gen_random_list(length=365, n_trades=10):
list_ = sorted([int(n) for n in
np.random.uniform(1, length, 2*n_trades)],
reverse=1)
list_.insert(0, 9999999)
return list_
def log(s, debug=1):
if debug:
print(s)
def print_result(*args, account=None, save=0):
py_fname = 'D:/algolab/random_walk.py'
b = ''
begin, end, day_num, lcopy = args
code = ', '.join([d.code for d in Hyzs.datalist])
name = ', '.join([d.name for d in Hyzs.datalist])
lcopy_str = ', '.join([str(e) for e in lcopy])
bj = '{:8.2f}'.format(account.capital)
pnl = '{:8.2f}'.format(account.cash)
nav = '{:8.2f}'.format(account.nav['nav'][-1])
r = ''
r += f'基于行业指数的, 随机漫步策略的回测分析报告{b}\n'
r += f' Date: 2023.12\n{b}\n\n'
r += f' {py_fname}\n'
r += f'数据集: {b}\n'
r += f' {code}\n'
r += f' {name}\n'
r += f' 时间段和自然日天数: {begin}--{end}, {day_num}天\n'
r+= f'策略参数:{b}\n'
r += f' 随机漫步开仓平仓时间线: [{lcopy_str}]\n'
r += f'结果:{b}\n'
r += f' 本金(capital): {bj}\n'
r += f' 盈亏(cash) : {pnl}\n'
r += f' 期末净值(nav) : {nav}\n'
r += ''
print(r)
return r
标签：plow,
py,
code6,
df,
self,
huo,
dt,
def
From： https://www.cnblogs.com/duan-qs/p/18000349