一、软件下载
https://epolestar.esunny.com.cn/download?ver=q
二、使用教程
hxxps://v.youku.com/v_show/id_XNjQwMTQ4ODI4OA
（一）添加策略
（二）
载入策略
主连合约选择：main
备注：选择回测期间那里可以选择“所有”，这样回测的时间就比较长。
三、问题
第一次回测的时候，发现它老是只显示今天和明天的，后来发现是因为用的系统默认的双均线策略，策略里面定死了调用k线的数量，所以是这样。
我用自己新建的策略就不会有这样的问题了。
如果要使界面添加合约生效，则屏蔽下面的代码：
SetBarInterval('ZCE|Z|SR|MAIN', 'M', 1, 4000, p2)
四、参数优化
添加参数优化的代码之后，要重新添加策略，否则相关代码不生效。
五、其他视频
https://www.bilibili.com/video/BV1PS2yBDEy7?spm_id_from=333.788.videopod.sections
六、策略
一、一个最简单的极智量化策略，用于获取白糖主力合约近20个交易日的收盘价，并通过LogInfo显示
import talib
def initialize(context):
# 设置白糖主力合约，日线周期，加载100根K线（确保足够获取20天数据）
SetBarInterval('ZCE|Z|SR|MAIN', 'D', 1, 100)
# 设置为K线触发模式
SetTriggerType(5)
def handle_data(context):
# 确保数据量足够（至少20根K线）
if CurrentBar() < 20:
return
# 获取收盘价序列
close_prices = Close()
# 取最近20个交易日的收盘价（-20到-1）
recent_20 = close_prices[-20:]
# 用LogInfo输出
LogInfo("=== 白糖主力合约最近20个交易日收盘价 ===")
for i, price in enumerate(recent_20):
LogInfo(f"第{i+1}天: {price}")
显示最近的1分钟k线的收盘价：
import talib
def initialize(context):
# 设置白糖主力合约，日线周期，加载100根K线（确保足够获取20天数据）
SetBarInterval('ZCE|Z|SR|MAIN', 'M', 1, 100)
# 设置为K线触发模式
SetTriggerType(5)
def handle_data(context):
# 确保数据量足够（至少20根K线）
if CurrentBar() < 20:
return
# 获取收盘价序列
close_prices = Close()
# 取最近20个交易日的收盘价（-20到-1）
recent_20 = close_prices[-2:-1] #如果是-1那根K线，就是会变化的，还没有走完的那根k线。
# 用LogInfo输出
LogInfo("=== 白糖主力合约前一根k线收盘价 ===")
for i, price in enumerate(recent_20):
LogInfo(f"第{i+1}根: {price}")
（二）双均线
import talib
import time
p1 = 5 # 快周期
p2 = 20 # 慢周期
qty = 1
def initialize(context):
# 设置K线稳定后发单
SetOrderWay(2)
# 设置基准合约，会覆盖界面设置的合约，建议通过界面设置(屏蔽SetBarInterval后则界面添加合约生效)
#SetBarInterval('SHFE|Z|RB|MAIN', 'D', 1, 3000, p2) 螺纹钢
#SetInitCapital(2*10000) 设置初资金为2万元
SetBarInterval('ZCE|Z|SR|MAIN', 'M', 1, 200, p2)
# 设置实盘运行
SetActual()
# 历史测回执行逻辑，定义一个函数用于回测。
def his_trigger(ma1, ma2): #ma1,ma2是一个序列
if ma1[-1] > ma2[-1] and MarketPosition() <= 0: #表示当前持仓
Buy(qty, Close()[-1])
elif ma1[-1] < ma2[-1] and MarketPosition() >= 0:
SellShort(qty, Close()[-1]) #建空仓，如果当前有多头持仓，该函数先平掉所有多仓，然后按照参数进行建空仓。
# 实盘阶段执行逻辑
def tim_trigger(ma1, ma2):
# 判断交易账号是否准备就绪 TODO
if ma1[-1] > ma2[-1] and A_TotalPosition() <= 0: #A_TotalPosition表示总持仓量，该持仓为所有持仓的合计值。MarketPosition 反映的是 “策略图表上”的理想信号仓位，用于回测；而 A_TotalPosition 反映的是 “真实账户”的持仓，用于实盘
prc = min(Q_AskPrice() + PriceTick(), Q_UpperLimit()) # 对盘超价 prc理解为price, Q_AskPrice() (卖一价)，Q_UpperLimit() (涨停价)
if A_TotalPosition() < 0:
A_SendOrder(Enum_Buy(), Enum_ExitToday(), A_SellPosition(), prc) # 平空仓 A_SendOrder(买卖方向, 开平仓类型, 委托手数, 委托价格)，Enum_Buy() 买入，Enum_ExitToday() 平今仓，A_SellPosition() 获取当前账户中所有的空头持仓手数
A_SendOrder(Enum_Buy(), Enum_Entry(), qty, prc) # 开多仓
elif ma1[-1] < ma2[-1] and A_TotalPosition() >= 0:
prc = max(Q_BidPrice() - PriceTick(), Q_LowLimit())
if A_TotalPosition() > 0:
A_SendOrder(Enum_Sell(), Enum_ExitToday(), A_BuyPosition(), prc) # 平多仓
A_SendOrder(Enum_Sell(), Enum_Entry(), qty, prc) # 开空仓
def handle_data(context):
if CurrentBar() < p2:
return;
ma1 = talib.MA(Close(), p1) #talib.MA(价格序列，计算周期，均线类型)， talib.MA() 的返回值是一个 NumPy 一维数组
ma2 = talib.MA(Close(), p2)
his = context.strategyStatus() == 'H' #获取当前策略状态，返回值'H'表示回测阶段， 'C'表示实时数据阶段
if his:
his_trigger(ma1, ma2)
else:
tim_trigger(ma1, ma2)
PlotNumeric("ma1", ma1[-1], 0xFF0000) #后面这些是画图
PlotNumeric("ma2", ma2[-1], 0x00aa00)
fit = NetProfit() + FloatProfit() - TradeCost() if his else A_CoverProfit() + A_ProfitLoss() - A_Cost()
PlotNumeric("fit", fit, 0x0000FF, False)
七、
请参考附件中的极智量化api手册，写一个广发证券LLT指标的极智量化策略，参考代码：
参数设置中设置：
CONTRACT = 'SHFE|Z|RB|MAIN' # 基准合约（螺纹钢主连，可修改）
def initialize(context):
SetBarInterval(CONTRACT , 'D', 1, '20150101')
SetInitCapital(2*10000)
SetSlippage(2)
SetActual()
八、回测期间
必须将下面的代码写到initialize里面，回测时选择全部k线，这个k线数量的设置才生效。
contract = Symbol()
SetBarInterval(contract, 'D', 1, 2100) # 使用所有可用历史数据
九、登陆账户
选择你的期货公司，然后账号和密码就是你在期货公司的账号和密码。
设置实盘交易。
自动运行
有自动登录选项，你设一下账户设定时间登录就行了。
十、实盘测试代码
import talib
def initialize(context):
# 设置白糖主力合约，日线周期，加载100根K线（确保足够获取20天数据）
SetBarInterval('ZCE|Z|SR|MAIN', 'M', 1, 100)
SetOrderWay(1)
SetActual()
def handle_data(context):
last_price = Q_Last()
open_price = Open()[-1]
if last_price > open_price and MarketPosition() <= 0:
Buy(1,Close()[-1])
elif last_price < open_price and MarketPosition() >= 0:
SellShort(2,Close()[-1])
十一、如何防止未成交单
十二、从极智量化中导出10条1分钟数据
########################系统示例不支持修改########################
# 布林带策略
# 根据布林带计算出高中低三个通道，较好的体现了行情的盘整和突破
# 根据行情时段与特点灵活运用会有不错的效果
##################################################################
import talib
import csv
import os
p = 8 # 计算周期
qty = 1 # 下单量
def initialize(context):
SetOrderWay(2)
# 关键修改：回测样本数设为 'A'（使用所有历史K线），barDataLen设为100000（足够大）
SetBarInterval('ZCE|Z|SR|MAIN', 'M', 1, 'A', 100000)
context.csv_filename = 'ohlc_1min_data.csv'
context.kline_data = [] # 用于存储所有K线数据
context.last_bar_index = -1 # 记录上一次处理的Bar索引，防止重复添加
LogInfo("策略初始化完成，将导出所有历史1分钟K线数据至：{}".format(context.csv_filename))
def handle_data(context):
# ========== 记录K线数据（仅回测阶段） ==========
if context.strategyStatus() == 'H':
cur_bar = CurrentBar()
if cur_bar != context.last_bar_index:
# 获取当前K线OHLC
dt = Date() # 格式YYYYMMDD
tm = Time() # 格式0.HHMMSS
# 拼接为字符串：YYYYMMDDHHMMSS
date_str = str(dt)
time_str = str(int(tm * 1000000)).zfill(6) # 转为HHMMSS
dt_stamp = date_str + time_str
open_p = Open()[-1]
close_p = Close()[-1]
low_p = Low()[-1]
high_p = High()[-1]
context.kline_data.append([dt_stamp, open_p, close_p, low_p, high_p])
context.last_bar_index = cur_bar
# ========== 原策略逻辑 ==========
if CurrentBar() < p:
return
upp, mid, low = talib.BBANDS(Close(), p, 2, 2)
if MarketPosition() <= 0 and Close()[-1] < low[-1]:
Buy(qty, Close()[-1])
elif MarketPosition() >= 0 and Close()[-1] > upp[-1]:
SellShort(qty, Close()[-1])
PlotNumeric('upp', upp[-1], RGB_Red())
PlotNumeric('mid', mid[-1], RGB_Blue())
PlotNumeric('low', low[-1], RGB_Green())
PlotNumeric("profit", NetProfit() + FloatProfit() - TradeCost(), 0xFF00FF, False)
def hisover_callback(context):
LogInfo("历史回测结束，开始导出1分钟K线数据至CSV...")
try:
if not context.kline_data:
LogWarn("无K线数据可导出")
return
file_path = os.path.join(os.getcwd(), context.csv_filename)
with open(file_path, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(['时间', '开盘价', '收盘价', '最低价', '最高价'])
for row in context.kline_data:
writer.writerow(row)
LogInfo("数据导出成功！共 {} 条记录，文件保存至：{}".format(len(context.kline_data), file_path))
except Exception as e:
LogError("导出CSV时发生异常：{}".format(str(e)))
def exit_callback(context):
LogInfo("策略退出")
导出后再用vnpy回测，以便核对效果：
from datetime import datetime
import pandas as pd
from vnpy_ctastrategy.backtesting import BacktestingEngine
from vnpy_ctastrategy.strategies.twenty03_strategy import Twenty03Strategy
from vnpy.trader.constant import Exchange, Interval
from vnpy.trader.object import BarData
def load_bars_from_csv(filepath: str, symbol: str, exchange: Exchange) -> list[BarData]:
"""从CSV文件加载K线数据，转换为BarData列表"""
# 读取CSV，自动检测分隔符（适用于制表符、逗号等）
df = pd.read_csv(filepath, sep=None, engine='python')
# 清理列名：去除空格和BOM头
df.columns = [col.strip().lstrip('\ufeff') for col in df.columns]
# 打印列名以便调试
print("CSV列名：", df.columns.tolist())
# 检查必须的列是否存在
required_cols = ['datetime', 'open', 'high', 'low', 'close', 'volume']
missing = [col for col in required_cols if col not in df.columns]
if missing:
raise KeyError(f"CSV缺少必要的列：{missing}")
bars: list[BarData] = []
for _, row in df.iterrows():
# 解析时间：尝试多种格式
dt_str = str(row['datetime']).strip()
try:
dt = datetime.strptime(dt_str, "%Y%m%d%H%M%S")
except ValueError:
# 如果时间格式不同，尝试其他常见格式
dt = pd.to_datetime(dt_str).to_pydatetime()
bar = BarData(
symbol=symbol,
exchange=exchange,
datetime=dt,
interval=Interval.MINUTE,
open_price=float(row['open']),
high_price=float(row['high']),
low_price=float(row['low']),
close_price=float(row['close']),
volume=float(row['volume']),
gateway_name="CSV"
)
bars.append(bar)
return bars
# 回测引擎设置
engine = BacktestingEngine()
engine.set_parameters(
vt_symbol="SR999.CZCE",
interval=Interval.MINUTE,
start=datetime(2018, 3, 23),
end=datetime(2018, 6, 23),
rate=0,
slippage=0,
size=10,
pricetick=0.2,
capital=1_000_000,
)
engine.add_strategy(Twenty03Strategy, {})
# 使用自定义CSV数据
engine.history_data = load_bars_from_csv(
"ohlc_1min_data.csv", "SR999", Exchange.CZCE
)
engine.run_backtesting()
df = engine.calculate_result()
engine.calculate_statistics()
engine.show_chart()
# 导出成交记录
trades = engine.get_all_trades()
df2 = pd.DataFrame([t.__dict__ for t in trades])
df2.to_csv('trade_log.csv', index=False)