一、基本思路
本框架的出发点是对真实交易过程的模仿，是一种仿真操作模型，包括了选股的过程，交易的过程，结算的过程，统计的过程等，是一个动态模型。
本框架的重点不在于交易策略，而在于框架的搭建。
本人Github仓库中有全部原码帮助有需要的人学习。本文的策略是基于大盘选股，而非单一股票的交易，可以通过修改策略函数来实现个股交易，但建议将本代码掌握清楚后才可尝试修改。
适合个人修改的代码框架，可以访问我的Github仓库
GitHub - Aegis1863/Quant_Trade_Frame: 量化交易，股票回测交易框架，仿真交易中的Solo_Stock_strategy.py文件
本模型包含的函数比较多，结构比较复杂，可以借助第三部分【策略思想和结构框架】部分来辅助理解。
二、数据来源
本框架的数据是基于Tushare的数据接口提取的，Tushare提供提取数据的接口代码示例，所用语言为Python，提取数据需要注册并获取token，此外需要注意积分权限。
注册Tushare大数据社区：
Tushare大数据社区本人Tushare ID：361998
Tushare数据接口截图：
![]()
日线接口
![]()
官方提供的接口代码示例
三、策略思想和结构框架
函数编写顺序并非执行顺序，请按照以下思路指引理解结构和功能。
- 本代码的策略是选择每天成交量最大的十只股票进行买入，该策略效果较差，仅作示范。交易股票首先要进行一系列初始化且准备数据，before_market_open(self) 函数执行此功能，其中会引用函数get_tick(self) 来获取数据，引用count_day(self) 函数计算需要交易的天数；初始化后根据函数strategy(self)里面所写策略进行选股，形成股票池
- 股票池传递给order_target_value(self) 进行交易，会计算手续费和总权益情况，其中会引用到交易函数trade(self) ，其中trade(self) 中会引用函数update_hold(self) 函数更新持仓情况。
- after_market_close(slef) 盘后函数功能仅为将日期往后推一日。
- 1~3步会在设定的时间内遍历，直到到达最终模拟日期。
- 结束交易后，statistics(self) 函数对最终结果进行统计，全部交易过程也将保留。
- picture_all(self) 函数以沪深300为基准对总权益进行绘图展示。
AstockTrading() 交易框架（类）
|- __init__() 初始化属性
|- get_tick() 获得行情数据
|- order_target_value() 将标的交易到指定仓位
|- before_market_open() 盘前初始化
|- after_market_close() 盘后函数
|- strategy() 策略函数
|- trade() 交易函数
|- update_hold() 更新持仓函数
|- picture_all() 作图函数
|- statistics() 统计函数
|- count_day() 用来计算起止日期中间有多少交易日，以此确定循环次数
|- run() 运行整个交易四、代码结构
引用的包
import tushare as ts
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime
import time定义类 class AstockTrading()，在__init__() 中设定属性。
诸如self._current_buy 均为类自身的属性，其值可以在类内部传递。
class AstockTrading(): # 策略类
def __init__(self, stratege_name, Start_date, End_date, origin_total_value):
token = '改成自己的token'
self._pro = ts.pro_api(token)
day = self._pro.trade_cal(exchange='', start_date=Start_date,
end_date=End_date)
self._current_buy = [] # 当天买入
self._current_sell = [] # 当天卖出
self._hands_list = [] # 当天买入手数
self.start_date = Start_date
self.end_date = End_date
self._datum_target = '399300.SZ' # 基准标的，沪深300
self._stratege_name = stratege_name
self._origin_total_value = origin_total_value # 持仓总市值
self._total_value = self._origin_total_value # 持仓初始市值
########## 主要属性 ↓
self._daily_tick = pd.DataFrame([]) # 初始化每日行情，在get_tick函数里面赋值
self._history_order = [] # 历史指令
self._calendar = day[day['is_open'] == 1].cal_date.apply(str)
# ↑ 交易日历,一次性赋值,列表,字符串
self._buy_list = [] # 每日买入股票列表
self._trade_number = 0 # 交易日期序号
self._history_value = [] # 历史市值
self._hold_stock = {} # 目前持仓股票
self._connot_sell_stock = [] # 跌停股，不能卖
self._base_rate = [] # 基准收益率
self._my_rate = [] # 个人收益率
self._new_calendar = pd.DataFrame([]) # 交易日历（作图用）
self._result = pd.DataFrame([]) # 最后用于统计盘前初始化
def before_market_open(self): # 盘前函数
self._connot_sell_stock = []
df2 = self._pro.suspend_d(suspend_type='S', \
trade_date=self._calendar.iloc[self._trade_number]).ts_code
self.get_tick() # 模拟交易前获取每日行情，相当于预先知道行情了，都是以收盘价交易
limit_down_list = self._pro.limit_list(
trade_date=self._calendar.iloc[self._trade_number],
limit_type='D').ts_code # 当天跌停股票
for stock in self._hold_stock: # 统计持仓里面哪些是跌停的
if stock in list(limit_down_list) or stock in list(df2):
self._connot_sell_stock.append(stock) # 持仓里面的跌停或停牌股列表计算交易日期数
def count_day(self): # 计算交易日期数
day = self._pro.trade_cal(exchange='', start_date=self.start_date,
end_date=self.end_date)
calendar = day[day['is_open'] == 1].cal_date.apply(str)
return len(calendar)获取行情数据
def get_tick(self): # 获取行情
daily_tick = self._pro.daily(
trade_date=self._calendar.iloc[self._trade_number]) # 原始每日行情数据
self._daily_tick = daily_tick.sort_values('amount', ascending=False,
inplace = False) # 按成交额排序策略选股
def strategy(self): # 策略重点修改这里
# 选股策略是选出每日成交量最大的10个股票，特殊情况顺延
df = self._daily_tick
# ↑为了便于书写，self._daily_tick赋值为当日行情数据df
self._buy_list = [] # 重新初始化买入列表
for index in list(df.index):
last_pchg = df.loc[index]['pct_chg']
code = df.loc[index]['ts_code']
if -3 <= last_pchg <= 3:
self._buy_list.append(code)
if len(self._buy_list) == 10 - len(self._connot_sell_stock):
break # 选到10支退出,排除跌停股，因此可能少于10个交易
def trade(self):# 选股结束，开始交易
df = self._daily_tick
MonPerStock = self._total_value / len(self._buy_list)
for stock in self._hold_stock: # stock 应当是代码字符串
if stock not in self._buy_list and stock not in self._connot_sell_stock:
# 如果已经持有的股票不在买入列表也没跌停或停牌，卖出
self.order_target_value(stock, 0) # 目标卖空
self._history_order.append([&#39;{}, {}, {}手, 现价{}元, 卖空&#39;.format(
df[df.ts_code==stock][&#39;trade_date&#39;].iloc[0],
stock,
&#39;all&#39;,
float(df[df.ts_code==stock][&#39;close&#39;].iloc[0]))]) # 增加记录
for stock in self._buy_list: # stock 应当是代码字符串
self.order_target_value(stock, MonPerStock) # 目标买入
self._history_order.append([&#39;{}, {}, {}手 现价{}元, 买入&#39;.format(
df[df.ts_code==stock][&#39;trade_date&#39;].iloc[0],
stock,
int(MonPerStock//float(df[df.ts_code==stock][&#39;close&#39;])//100),
float(df[df.ts_code==stock][&#39;close&#39;].iloc[0]))]) # 增加记录
self.update_hold() # 更新持仓和市值
self.after_market_close() # 盘后，日期加一更新持仓
def update_hold(self): # 更新持仓和市值
df = self._daily_tick
for num in range(len(self._current_sell)):
del self._hold_stock[self._current_sell[num]]
for num in range(len(self._current_buy)):
self._hold_stock[self._current_buy[num]] = \
[int(self._hands_list[num]),
float(df[df.ts_code==self._current_buy[num]][&#39;close&#39;])]
self._current_buy = [] # 当天买入
self._current_sell = [] # 当天卖出
self._hands_list = [] # 当天买入手数
self._history_value.append(self._total_value) # 记录历史市值盘后函数
用于处理交易结束收盘后的剩余工作
def after_market_close(self): # 盘后函数
self._trade_number += 1 # 今日行情获取完毕之后，序号加1，下次获取下一日行情数据统计函数
用于对交易进行统计
def statistics(self): # 统计结果
# 把历史市值改成pd.Series形式
plt.rcParams[&#39;font.sans-serif&#39;] = [&#39;Microsoft YaHei&#39;]
self._new_calendar[&#39;trade_date&#39;] = [datetime.strptime(day, &#39;%Y%m%d&#39;).date()\
for day in self._calendar]
hs300 = self._pro.index_daily(ts_code=self._datum_target,
start_date=self.start_date,
end_date=self.end_date)
hs300_close = list(hs300[&#39;close&#39;][::-1]) # 沪深300收盘价
for i in range(len(hs300_close)):
self._base_rate.append((hs300_close - hs300_close[0])/hs300_close[0])
# ↑ 沪深300累计收益率
for i in range(len(self._history_value)):
self._my_rate.append((self._history_value - \
self._origin_total_value)/self._origin_total_value)绘图函数
对权益变动情况进行绘制
def picture_all(self): # 交易结束,作图和结算
print(&#39;>> 最终市值{:.3f}元，收益率{:.3f}%\n&#39;.format(self._total_value,
(self._total_value - \
self._origin_total_value)/\
self._origin_total_value*100))
self._result[&#39;历史市值&#39;] = self._my_rate
self._result[&#39;沪深300&#39;] = self._base_rate
self._result[&#39;超额收益&#39;] = self._result[&#39;历史市值&#39;] - self._result[&#39;沪深300&#39;]
plt.plot(self._new_calendar, self._result)
plt.legend([&#39;历史市值&#39;, &#39;沪深300&#39;, &#39;超额收益&#39;])
plt.xlabel(&#39;日期&#39;,fontsize=&#39;15&#39;)
plt.ylabel(&#39;收益率&#39;,fontsize=&#39;15&#39;)
plt.title(&#39;{}策略&#39;.format(self._stratege_name),fontsize=15)
plt.grid()
plt.show()运行函数
用于跑全部代码
def run(self): # 运行全部交易流程
days = self.count_day()
for i in range(days): # 循环若干交易日
time.sleep(0.05)
self.before_market_open()
self.strategy()
self.trade()
if (i+1)%5 == 0:
print(&#39;>> 已完成{}天交易，还剩下{}天&#39;.format(i+1,days-i-1)) # 通报进度
print(&#39;## 交易完毕\n&#39;)
self.statistics()
self.picture_all() # 统计作图用户调用方法
mys = AstockTrading(&#39;amount_strategy&#39;, &#39;20190101&#39;, &#39;20190531&#39;, 1000000)
# 设定策略名称，交易起止日期，本金
mys.run() # 运行
print(&#39;利用以下命令查看相关信息：\n>> mys.picture_all() 显示图像&#39;)
print(&#39;>> mys._history_order 显示历史指令&#39;)最后
由于本文所调用的数据在Tushare中所需积分比较高，因此直接调用可能会获取数据失败，请先对照规则查看自己的积分是否足够。
本代码经测试，在积分权限正常的情况下，可以正常运行并且输出数据。
由于本人能力有限，可能存在一些代码冗余的问题，欢迎指正，本代码整体较为复杂，初学者可以学习思路和方法自己尝试编写框架。 |
转播
