# 导入相应包
from biglearning.api import M
import math
from bigtrader.sdk import *
import pandas as pd
import warnings
warnings.filterwarnings('ignore')
def initialize(context):
"""策略初始化函数，只触发一次。可以在该函数中初始化一些变量，如读取配置和全局使用数据"""
#输出关键日志
msg = "initialize:"
context.write_log(msg, stdout=1)
context.ins = context.instruments #从传入参数中获取需要交易的合约
context.short_ma = 5 #均线短周期参数
context.long_ma = 64 #均线长周期参数
context.set_stock_t1(False) # False表明可以T0交易
def before_trading_start(context, data):
"""盘前处理，策略盘前交易函数，每日盘前触发一次。可以在该函数中一些启动前的准备，如订阅行情等"""
#输出关键日志
msg = "before_trading dt:{}".format(data.current_dt)
context.write_log(msg, stdout=1) # stdout=1表示在console里打印
#如果需要处理tick数据，需要添加tick处理函数:handle_tick(context, tick):
#如果需要使用handle_data，需要添加处理函数:handle_data(context, data):
def handle_data(context, data):
"""行情通知函数"""
#获取当前时间
cur_date = data.current_dt # 当前分钟k线时间
cur_hm = cur_date.strftime('%H:%M')
# 获取账户可用资金
total_portfolio = context.portfolio.cash
for instr in context.ins:
#获取持仓情况
position = context.get_position(instr)
#最新价格
price = data.current(instr, "close")
#获取1m历史数据数据
hist = context.history_data(instr, ["open","high","low","close"], context.long_ma+5, "1m")
# 计算均值
short_ma = pd.Series(hist['close']).rolling(context.short_ma).mean()
long_ma = pd.Series(hist['close']).rolling(context.long_ma).mean()
# 数据 条数不够就返回
if len(long_ma) < context.long_ma:
return
# 等权重
weight = 1/len(context.ins)
#短周期均线上穿越长周期均线买入(这里的判断是当前的k线上上穿，上一根k线没有达到上穿条件)
if short_ma.iloc[-2]< long_ma.iloc[-2] and short_ma.iloc[-1]> long_ma.iloc[-1]:
#当前没有持仓则市价买入
if (position.current_qty == 0):
#计算买入此股票的数量，不要超过总资金的某个比例
order_num = math.floor(total_portfolio*weight/price/100)*100
rv = context.order(instr, order_num, price, order_type=OrderType.MARKET)
msg = "{} 买入{} 最新价={:.2f} 下单函数返回={}".format(cur_date,instr,price,rv)
context.write_log(msg, stdout=0)
#短周期均线下穿越长周期均线卖出
elif short_ma.iloc[-2] > long_ma.iloc[-2] and short_ma.iloc[-1] < long_ma.iloc[-1]:
#有持仓则卖出
if (position.current_qty != 0):
rv = context.order(instr, -position.current_qty, price, order_type=OrderType.MARKET)
msg = "{} 卖出{} 最新价={:.2f} 下单函数返回={}".format(cur_date,instr,price,rv)
context.write_log(msg, stdout=0)
def handle_order(context, order):
"""委托回报通知函数，每个订单状态有变化时会触发"""
#输出关键日志
msg = "handle_order data:{}".format(order.log_str())
context.write_log(msg, stdout=0) # stdout=0表示不在console里打印
def handle_trade(context, trade):
"""成交回报通知函数，有成交时会触发"""
#输出关键日志
msg = "handle_trade data:{}".format(trade.log_str())
context.write_log(msg, stdout=0)
backtest_result = M.hftrade.v2(
instruments=['688517.SHA'],
start_date='2023-01-01',
end_date='2023-02-09',
handle_data=handle_data,
before_trading_start=before_trading_start,
initialize=initialize,
volume_limit=0.025,
order_price_field_buy='open',
order_price_field_sell='open',
capital_base=1000000,
frequency='minute',
price_type='真实价格',
product_type='股票',
plot_charts=True,
backtest_only=False,
benchmark='000300.HIX' # 比较基准
)