★ FEATURED ARTICLE 简介这份资源是面向Python开发者与量化交易初学者的期货程序化交易系统完整项目源码基于Python实现适合用于毕业设计、课程实践或量化策略入门学习。项目采用前后端分离结构后端以Python业务逻辑与数据库脚本为核心前端由Vue组件构建可视化界面并配套配置文件与初始化脚本方便读者理解一套交易系统从建库、建表到启动运行的完整链路。压缩包共630个文件约28.24MB包含159个svg图标、129个vue组件、92张jpg图片、58个js脚本、46个py源码及39个png素材另有css、bat批处理、json配置、sql建表脚本与字体文件等覆盖界面资源、业务代码与部署脚本多个层面。目前已有31人学习下载。读者可从中获取可运行的期货交易系统源码、数据库初始化方案、前后端交互示例与配置模板并借助批处理脚本快速完成环境搭建与启动适合作为量化交易方向的学习参考与二次开发基础。1. 从一份毕设标题说起Python 期货程序化交易系统到底在做什么很多人第一次看到「基于 Python 的期货程序化交易系统」这类标题第一反应是「又一个毕设」。但如果你真的在期货公司待过、或者自己盯过夜盘就会知道这件事的门槛不在写代码而在把行情、策略、下单、风控这四件事串成一条不会断的链路。我见过太多人用 Python 写了个双均线回测曲线漂亮得不行一上模拟盘就发现滑点吃掉全部利润或者行情断线后策略还在拿旧价格反复挂单。这套系统的核心价值就是把这些散落的环节收进一个可运行、可监控、可复盘的工程里。它适合两类人一类是想把交易想法快速验证的量化新手另一类是需要一套能自己改、不依赖黑匣子平台的开发者。你不需要先成为期货专家但得接受一个现实——策略只是系统里最小的一块真正花时间的是数据接入、订单状态管理和异常处理。下面我按自己搭过的一套结构从环境到落地讲清楚哪些地方能抄哪些地方必须自己踩一遍。2. 系统骨架怎么搭行情、策略、订单、风控四层拆开看2.1 为什么不能把策略和下单写在一个文件里新手最容易犯的错是在一个main.py里while True拉行情、算信号、直接调下单接口。跑起来没问题但一旦要加第二个策略、要换交易所、要做盘中风控整个文件就变成一团。我一般会拆成四层数据层负责行情和 K 线的获取与缓存策略层只接收标准化后的行情并输出目标仓位执行层把目标仓位翻译成具体报单并跟踪成交风控层独立于策略做仓位上限、单笔最大亏损和频率限制。这样拆的好处是策略层可以完全不知道用的是哪家柜台接口回测和实盘共用同一份策略代码。代价是前期要多写一些胶水代码但后面换接口、加策略时省下的时间远超这点投入。常见做法是用一个事件循环把四层串起来行情到达触发策略计算策略输出信号进入执行队列执行结果再回调给风控和记录模块。2.2 用 Python 把最小可运行骨架跑起来下面这段代码是一个不依赖任何柜台接口的最小骨架用模拟行情驱动重点看结构而不是功能。你可以直接复制运行观察事件是怎么在四层之间流动的。import time import random from collections import deque from dataclasses import dataclass, field from typing import Optional dataclass class Tick: symbol: str last_price: float timestamp: float dataclass class Order: symbol: str direction: str # buy or sell volume: int price: Optional[float] None status: str pending # pending / filled / rejected class DataFeed: 模拟行情源实盘时替换为柜台或行情 API 的订阅回调 def __init__(self, symbols): self.symbols symbols self.price {s: 3000.0 for s in symbols} def poll(self): for s in self.symbols: # 模拟价格随机游走实盘这里换成真实 tick self.price[s] random.uniform(-2, 2) yield Tick(s, round(self.price[s], 1), time.time()) class Strategy: 策略层只输出目标仓位不关心怎么下单 def __init__(self, symbol, window20): self.symbol symbol self.window window self.history deque(maxlenwindow) def on_tick(self, tick: Tick) - Optional[str]: self.history.append(tick.last_price) if len(self.history) self.window: return None ma sum(self.history) / len(self.history) # 价格上穿均线做多下穿做空简单示例 if tick.last_price ma * 1.002: return buy if tick.last_price ma * 0.998: return sell return None class RiskManager: 风控层独立于策略做仓位和频率限制 def __init__(self, max_position5, max_orders_per_min10): self.max_position max_position self.max_orders_per_min max_orders_per_min self.position 0 self.order_times deque() def check(self, direction: str) - bool: now time.time() while self.order_times and now - self.order_times[0] 60: self.order_times.popleft() if len(self.order_times) self.max_orders_per_min: return False if direction buy and self.position self.max_position: return False if direction sell and self.position -self.max_position: return False return True def update(self, direction: str, volume: int): self.position volume if direction buy else -volume self.order_times.append(time.time()) class Executor: 执行层把信号翻译成报单实盘时对接柜台下单接口 def __init__(self, risk: RiskManager): self.risk risk self.orders [] def send(self, symbol: str, direction: str, volume: int 1): if not self.risk.check(direction): print(f[风控拦截] {symbol} {direction} 被拒绝) return order Order(symbol, direction, volume) # 模拟立即成交实盘这里要处理部分成交和撤单 order.status filled self.orders.append(order) self.risk.update(direction, volume) print(f[成交] {symbol} {direction} {volume}手 当前仓位{self.risk.position}) if __name__ __main__: feed DataFeed([rb2310]) strategy Strategy(rb2310) risk RiskManager(max_position5, max_orders_per_min10) executor Executor(risk) for i in range(200): for tick in feed.poll(): signal strategy.on_tick(tick) if signal: executor.send(tick.symbol, signal) time.sleep(0.01)这段代码里DataFeed.poll是唯一需要替换成真实行情的地方实盘时改成柜台 SDK 的订阅回调即可。Strategy.on_tick只返回方向字符串不碰任何下单逻辑这样回测时你可以用历史数据直接喂给它。RiskManager里的max_position和max_orders_per_min是最该先调的两个参数前者防止策略失控加仓后者防止行情抖动时疯狂报单。Executor.send里我故意把成交写成立即完成实盘要处理挂单、部分成交和撤单回报这是后面避坑章节要展开的点。2.3 回测和实盘共用策略代码的关键约束要让同一份策略代码在回测和实盘都能跑策略层必须遵守两条约束第一只依赖传入的 tick 或 bar不主动去拉数据第二不假设成交一定发生信号和成交之间要允许有延迟。我见过有人策略里直接写if 价格 均线: 买入()回测时因为成交价就是当前价所以曲线好看实盘一挂单价格就跑了。正确做法是策略只输出「目标仓位」由执行层决定用什么价格、什么时候报单。回测引擎通常按 bar 的收盘价撮合实盘则是报单后等成交回报。这两者的差异会导致同样的策略在回测和实盘上表现不同尤其是用限价单的时候。我的习惯是在回测里加一个滑点参数比如每次成交价按方向偏移 1 到 2 个最小变动价位这样回测结果更接近实盘。这个参数没有标准值螺纹钢主力合约我一般设 1 个点流动性差的品种要设更大。3. 行情接入与数据清洗从柜台 API 到本地 K 线的落地细节3.1 行情接口选型CTP、易盛还是本地模拟国内期货程序化交易最常见的行情来源是 CTP 接口它提供 tick 级别的行情推送覆盖国内主要交易所。选它的理由是稳定、文档相对齐全、社区资料多。易盛接口在部分品种和席位上延迟更低但接入成本更高。如果你只是做策略验证不需要真实行情用本地模拟数据或者历史 tick 回放就够了没必要一上来就接实盘。接入 CTP 行情一般用 Python 的封装库常见做法是通过一个行情前置地址登录订阅合约后在一个回调函数里接收 tick。这里有个容易忽略的点CTP 的行情回调是在网络线程里触发的如果你在回调里直接做策略计算和下单会阻塞行情接收导致丢包。我一般会在回调里只做一件事——把 tick 塞进一个线程安全的队列然后由主循环从队列里取数据做后续处理。import queue import threading tick_queue queue.Queue(maxsize10000) def on_rtn_depth(market_data): CTP 行情回调只做入队不做任何计算 try: tick_queue.put_nowait({ symbol: market_data.InstrumentID, last: market_data.LastPrice, bid: market_data.BidPrice1, ask: market_data.AskPrice1, volume: market_data.Volume, time: market_data.UpdateTime, }) except queue.Full: # 队列满说明消费太慢丢弃最旧数据并告警 print([告警] tick 队列满丢弃数据) def consume_loop(): 主循环从队列取数据做策略和风控 while True: tick tick_queue.get() # 这里调用策略层和执行层 process_tick(tick)queue.Queue(maxsize10000)的容量要根据行情频率设螺纹钢主力合约活跃时每秒可能有几十个 tick10000 够缓冲几分钟。put_nowait配合queue.Full异常处理保证行情回调永远不会被阻塞。消费循环里如果处理太慢队列会满这时候丢数据比阻塞行情线程更安全但一定要打日志告警否则你根本不知道策略是在用旧数据做决策。3.2 把 tick 合成 K 线的三个边界坑策略通常需要 K 线而不是裸 tick合成 K 线看起来简单实际有三个坑。第一是时间边界期货有夜盘夜盘和日盘之间怎么切分 K 线不同交易所规则不同。第二是集合竞价开盘前那几分钟的 tick 要不要算进第一根 K 线。第三是收盘最后一秒的 tick如果没处理好最后一根 K 线会缺数据。我的做法是用一个BarGenerator类按固定时间窗口聚合窗口结束时强制输出当前 bar不管有没有新 tick。下面是一个简化实现from datetime import datetime, timedelta class BarGenerator: def __init__(self, symbol, interval_seconds60): self.symbol symbol self.interval interval_seconds self.current_bar None self.bar_start None def on_tick(self, tick): ts datetime.fromtimestamp(tick[timestamp]) # 计算当前 tick 属于哪个 bar 窗口 window_start ts.replace(second0, microsecond0) if self.bar_start is None: self.bar_start window_start self.current_bar self._new_bar(tick) return None if window_start self.bar_start: # 窗口切换输出上一根 bar finished self.current_bar self.bar_start window_start self.current_bar self._new_bar(tick) return finished else: self._update_bar(self.current_bar, tick) return None def _new_bar(self, tick): return { symbol: self.symbol, open: tick[last], high: tick[last], low: tick[last], close: tick[last], volume: tick[volume], start: self.bar_start, } def _update_bar(self, bar, tick): bar[high] max(bar[high], tick[last]) bar[low] min(bar[low], tick[last]) bar[close] tick[last] bar[volume] tick[volume] # 注意这里应该是增量不是累计_update_bar里volume直接赋值是错的实盘 tick 的Volume字段通常是当日累计成交量要算增量得用当前值减去上一笔的值。这个坑我踩过回测时成交量看起来正常实盘发现每根 K 线的成交量都是当天累计值策略里用成交量做的过滤条件全部失效。正确做法是在BarGenerator里记录上一笔的累计成交量每次算差值。另外窗口切换的判断用window_start self.bar_start在跨日时会出问题因为replace(second0)不会处理日期跳变。更稳妥的做法是用时间戳整除窗口长度比如int(ts.timestamp()) // self.interval * self.interval这样跨日跨月都不会错。4. 策略落地与订单管理从信号到成交的完整链路4.1 一个可实盘的双均线策略怎么写才不翻车双均线是最常见的入门策略但直接写「短均线上穿长均线就买」在实盘里会翻车原因是信号在均线附近反复穿越时会产生大量无效报单。我一般会加两个过滤条件一是要求穿越幅度超过一定阈值比如短均线超过长均线 0.2% 才触发二是加一个最小持仓时间避免刚开仓就反向平仓。class DualMAStrategy: def __init__(self, symbol, fast5, slow20, threshold0.002, min_hold_seconds300): self.symbol symbol self.fast fast self.slow slow self.threshold threshold self.min_hold_seconds min_hold_seconds self.fast_prices deque(maxlenfast) self.slow_prices deque(maxlenslow) self.position 0 self.last_trade_time 0 def on_bar(self, bar): self.fast_prices.append(bar[close]) self.slow_prices.append(bar[close]) if len(self.slow_prices) self.slow: return None fast_ma sum(self.fast_prices) / len(self.fast_prices) slow_ma sum(self.slow_prices) / len(self.slow_prices) now bar[start].timestamp() # 最小持仓时间过滤 if now - self.last_trade_time self.min_hold_seconds: return None spread (fast_ma - slow_ma) / slow_ma if spread self.threshold and self.position 0: self.last_trade_time now return buy if spread -self.threshold and self.position 0: self.last_trade_time now return sell return Nonethreshold设 0.002 意味着短均线要比长均线高 0.2% 才买这个值太小过滤不掉震荡太大会错过趋势启动。螺纹钢上我一般用 0.001 到 0.003 之间具体要看品种波动率。min_hold_seconds设 300 秒是防止在均线附近反复打脸但如果你做的是日内高频这个值要调小甚至去掉。注意position字段需要和执行层的实际仓位同步不能只靠策略自己记否则手动干预后策略状态就乱了。4.2 订单状态机挂单、部分成交、撤单怎么管实盘下单最麻烦的不是发单而是跟踪订单状态。CTP 的报单回报有「已提交」「部分成交」「全部成交」「已撤单」「报单错误」等多种状态而且回报顺序不保证。我见过有人收到「全部成交」后还在等「已提交」结果订单状态永远卡住。正确做法是用一个字典按OrderRef或OrderSysID索引订单每次回报更新对应订单的状态状态只能单向推进。class OrderTracker: def __init__(self): self.orders {} # order_ref - order dict def on_order_return(self, order_ref, status, filled_volume, remaining): if order_ref not in self.orders: # 收到未知订单回报可能是重启后丢失了本地状态 print(f[告警] 未知订单回报 {order_ref}) return order self.orders[order_ref] # 状态只能向前推进防止乱序回报覆盖 status_rank {pending: 0, partial: 1, filled: 2, cancelled: 2, rejected: 2} if status_rank.get(status, 0) status_rank.get(order[status], 0): return order[status] status order[filled] filled_volume order[remaining] remaining if status in (filled, cancelled, rejected): # 终态可以清理或归档 self._finalize(order) def _finalize(self, order): print(f[订单终态] {order[order_ref]} {order[status]} 成交{order[filled]}手)status_rank这个映射是关键它保证即使回报乱序订单状态也不会从「已成交」退回「部分成交」。on_order_return里对未知订单的处理也很重要实盘程序重启后本地订单字典是空的但柜台可能还有挂单这时候收到回报不能直接忽略要主动去查询柜台持仓和挂单做对账。我一般会在程序启动时先查一遍柜台挂单把本地状态同步过来再开始接收新回报。4.3 用配置文件管理参数别把参数写死在代码里策略参数、风控阈值、合约代码这些经常要改的东西千万别写死在.py文件里。我习惯用一个 YAML 或 JSON 配置文件程序启动时加载改参数不用重新部署代码。下面是一个配置示例# config.yaml data: source: ctp # ctp / simulation symbols: [rb2310, i2310] bar_interval: 60 # 秒 strategy: name: dual_ma params: fast: 5 slow: 20 threshold: 0.002 min_hold_seconds: 300 risk: max_position: 5 max_orders_per_min: 10 max_daily_loss: 5000 # 当日最大亏损触发后停止交易 executor: order_type: limit # limit / market slippage_ticks: 1加载配置的代码很简单但要注意类型转换和默认值处理。max_daily_loss这个参数我强烈建议加上实盘最怕的就是策略在极端行情里连续亏损有个硬止损能救命。触发后不是简单停止策略而是要撤销所有挂单、平掉所有仓位然后等人工确认后再重启。5. 避坑与排查实盘跑起来才会遇到的五个问题5.1 行情断线后策略还在用旧价格挂单现象网络抖动导致行情推送中断但策略循环还在跑用最后一笔 tick 的价格反复计算信号并报单结果在错误价格上成交。原因行情回调和策略循环是解耦的行情断了策略不知道。解决在DataFeed里加一个心跳检测超过 N 秒没收到任何 tick 就设置一个stale标志策略层看到这个标志就跳过计算执行层同时撤销所有挂单。N 一般设 5 到 10 秒夜盘流动性差的时候可以放宽。5.2 报单被柜台拒绝但程序没处理现象下单后没有成交回报查柜台发现报单被拒绝原因是保证金不足或合约代码错误。原因on_order_return里只处理了成交和撤单没处理「报单错误」状态。解决在订单状态机里把rejected当作终态处理同时打印错误码和错误信息。CTP 的错误码文档要常备比如「资金不足」「仓位不足」「合约不存在」这些常见错误要能快速定位。我一般会在执行层加一个重试逻辑但只对「网络超时」这类临时错误重试对「资金不足」这种硬错误直接告警停止。5.3 回测赚钱实盘亏钱滑点和手续费没算对现象回测年化 30%实盘跑一个月亏 5%。原因回测里按收盘价成交没算滑点和手续费实盘限价单经常挂不上改成市价单后滑点吃掉利润。解决回测里加滑点参数按品种设不同值螺纹钢 1 个点、铁矿 0.5 个点、股指 0.2 个点。手续费按交易所标准算注意平今仓手续费可能比开仓高好几倍。另外回测的成交假设要保守限价单不保证成交市价单按对手价加滑点成交。5.4 策略参数在样本内表现好样本外一塌糊涂现象用过去一年数据优化出来的参数在最近三个月完全失效。原因过拟合参数在历史数据上被过度优化。解决把数据分成训练集和验证集训练集上找参数验证集上确认。如果验证集表现明显差于训练集说明过拟合。更稳妥的做法是用滚动窗口每三个月重新优化一次参数但要注意参数切换时的仓位处理。我一般会限制参数搜索范围比如均线周期不超过 60避免找到一些在逻辑上说不通的参数。5.5 程序崩溃后仓位状态丢失现象程序异常退出重启后不知道当前持仓和挂单策略按空仓开始交易结果重复开仓。原因本地状态没有持久化。解决每次订单状态变化和成交后把持仓、挂单、当日盈亏写到一个本地文件或 SQLite 数据库。程序启动时先读本地状态再查柜台状态做对账不一致时以柜台为准。这个对账逻辑一定要写我见过有人因为没对账程序重启后重复开了三倍仓位幸好发现得早。6. 进阶技巧用事件回放做策略验证和参数调优实盘跑稳之后下一步是怎么快速验证策略改动。直接上模拟盘太慢一天只能跑一次。我的做法是录一段真实 tick 数据存成文件然后用事件回放的方式在本地重放。这样一天的数据可以在几分钟内跑完参数调优效率高很多。录制 tick 的代码很简单在行情回调里加一个写文件逻辑import json class TickRecorder: def __init__(self, filepath): self.filepath filepath self.file open(filepath, a, encodingutf-8) def on_tick(self, tick): # 每行一个 JSON方便后续按行读取 self.file.write(json.dumps(tick, ensure_asciiFalse) \n) self.file.flush() # 防止程序崩溃丢数据 def close(self): self.file.close()回放的时候按时间戳排序用一个模拟时钟驱动而不是time.sleep真实等待。这样回放速度可以任意快但要注意策略里如果有依赖真实时间的逻辑比如min_hold_seconds要用模拟时钟而不是time.time()。我一般会在策略里注入一个clock对象回测时用模拟时钟实盘时用真实时钟。回放验证的重点不是看收益曲线而是看三件事第一订单状态流转是否正常有没有卡在中间状态的订单第二风控是否按预期拦截特别是仓位上限和频率限制第三异常处理是否触发比如行情断线、报单被拒。这三件事在实盘里出问题代价很高在回放里发现就是赚到。参数调优我一般用网格搜索但不会全量搜而是先按逻辑确定合理范围再在范围内采样。比如双均线的fast取 3 到 10slow取 15 到 40threshold取 0.001 到 0.005组合数控制在几百个以内。每个组合跑一遍回放记录夏普比率、最大回撤和交易次数。交易次数太少的组合直接排除因为统计上不可靠。最后说一个我自己的习惯每次策略改动上线前先用回放跑一遍最近一周的数据确认没有异常订单和风控触发再上模拟盘跑一天最后才切实盘。这个流程看起来慢但比直接上实盘然后半夜被报警叫醒要快得多。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
提前堵住了它)
