简介这是一份从策略为王论坛通过SVN获取的开源交易软件源代码面向对证券行情系统、客户端开发感兴趣的开发者与量化爱好者可用于学习行情展示、K线绘制与实时走势图等核心功能的实现思路。压缩包共1299个文件约10.54MB以219个h头文件、215个cpp源文件为主体配合11个vcproj工程文件与2个sln解决方案另有bmp、ico等界面资源及lib、def等库文件整体结构接近完整工程。代码已用VS2008编译通过并可正常运行行情、K线、实时走势图等常见交易软件功能基本齐备。目前已有1140人学习下载。作者同时希望获得推送行情的服务器程序源码以及行情服务器搭建、推送服务编写与接收对应方式的说明因此该资源也适合作为研究行情服务端与客户端通信的起点便于读者在此基础上补全服务端链路、理解数据推送与接收的对应关系。1. 策略为王开源交易软件源代码一套能跑起来的量化框架长什么样很多人第一次接触量化交易都是从“找一套开源交易软件源代码”开始的。搜出来的东西五花八门有回测框架、有实盘接口封装、有策略信号库还有一堆跑不起来的半成品。真正能用的开源交易软件源代码核心不在界面多漂亮而在于策略层能不能被清晰地拆出来、独立验证、再挂到实盘上。策略为王这句话不是口号它意味着整套代码的组织方式必须围绕策略的生命周期来设计策略怎么写、参数怎么传、回测怎么跑、结果怎么评估、实盘怎么接。如果你是想从零搭一套自己的量化系统或者想读懂一套开源框架的骨架这篇笔记会按“策略定义→数据接入→回测引擎→实盘对接→避坑”的顺序把每个环节的代码和参数讲清楚。适合有 Python 基础、想认真做策略而不是玩票的开发者。2. 策略层怎么设计从策略模式到可插拔的信号函数2.1 为什么开源交易软件都爱用策略模式翻看主流开源交易软件源代码你会发现策略层几乎都用了策略模式Strategy Pattern。这不是为了炫技而是因为交易策略的变体太多了均线交叉、MACD 双底、布林带突破、网格、动量轮动每一种的入场出场逻辑都不一样但它们的生命周期是一样的——初始化、接收行情、判断信号、下单、记录状态。如果把这些逻辑全写在一个类里改一个策略就要动核心代码回测和实盘还会互相污染。策略模式的做法是定义一个抽象基类把“怎么算信号”交给子类实现框架只负责调用。这样新增策略就是新增一个文件不用碰引擎。下面是一个最小可用的策略基类我一般会把它放在strategies/base.pyfrom abc import ABC, abstractmethod class BaseStrategy(ABC): def __init__(self, params: dict): # params 里放策略参数比如均线周期、止损比例 self.params params self.position 0 # 当前持仓正数多头负数空头 self.entry_price None # 开仓均价用于计算盈亏 abstractmethod def on_bar(self, bar: dict) - dict: 每根 K 线调用一次返回交易指令 bar 格式: {datetime:..., open:..., high:..., low:..., close:..., volume:...} 返回: {action: buy|sell|hold, size: float, reason: str} pass def update_position(self, action: str, price: float, size: float): # 成交后更新持仓状态回测和实盘共用 if action buy: self.position size self.entry_price price if self.entry_price is None else \ (self.entry_price * (self.position - size) price * size) / self.position elif action sell: self.position - size if self.position 0: self.entry_price None这段代码的关键点有三个。第一on_bar是唯一需要子类实现的接口框架不关心你用什么指标只关心你返回什么指令。第二update_position放在基类里保证回测和实盘对持仓的计算逻辑完全一致避免“回测赚钱实盘亏”的经典翻车。第三params用字典传入方便从配置文件或命令行覆盖不用改代码就能调参。参数说明size建议用“股数/币数”而不是“金额”因为不同标的的最小交易单位不同金额换算放在执行层做更安全。reason字段看起来多余但实盘排查时它是你的黑匣子一定要填。2.2 用 MACD 双底写一个可回测的策略子类有了基类写一个具体策略就是填on_bar。以热搜里常出现的“MACD 双底”为例逻辑是价格创新低但 MACD 柱状线不创新低形成底背离第二次金叉时买入。下面是一个简化但可运行的实现import pandas as pd class MACDDoubleBottom(BaseStrategy): def __init__(self, params): super().__init__(params) self.closes [] # 缓存收盘价用于计算 MACD self.macd_hist [] # 缓存 MACD 柱 self.last_low_price None self.last_low_hist None self.golden_cross False def _calc_macd(self): # 用 pandas 计算 MACD参数从 params 取默认 12/26/9 s pd.Series(self.closes) fast s.ewm(spanself.params.get(fast, 12), adjustFalse).mean() slow s.ewm(spanself.params.get(slow, 26), adjustFalse).mean() dif fast - slow dea dif.ewm(spanself.params.get(signal, 9), adjustFalse).mean() hist (dif - dea) * 2 return dif.iloc[-1], dea.iloc[-1], hist.iloc[-1] def on_bar(self, bar): self.closes.append(bar[close]) if len(self.closes) self.params.get(slow, 26) 10: return {action: hold, size: 0, reason: warmup} dif, dea, hist self._calc_macd() self.macd_hist.append(hist) # 记录价格低点和对应的 MACD 柱低点 if len(self.macd_hist) 1 and hist self.macd_hist[-2]: if self.last_low_price is None or bar[close] self.last_low_price: self.last_low_price bar[close] self.last_low_hist hist # 底背离判断价格更低但 MACD 柱更高 divergence (self.last_low_price is not None and bar[close] self.last_low_price and hist self.last_low_hist) # 金叉判断 prev_dif dif - (hist / 2) # 近似上一根 DIF实盘建议缓存 if dif dea and prev_dif dea: self.golden_cross True if divergence and self.golden_cross and self.position 0: self.golden_cross False return {action: buy, size: self.params.get(size, 100), reason: MACD double bottom divergence} # 止损亏损超过阈值就平仓 if self.position 0 and self.entry_price: loss_pct (bar[close] - self.entry_price) / self.entry_price if loss_pct -self.params.get(stop_loss, 0.05): return {action: sell, size: abs(self.position), reason: stop loss} return {action: hold, size: 0, reason: }逻辑说明先缓存足够多的收盘价再开始计算避免前几根 K 线指标失真。底背离的判断依赖两个低点代码里用“价格创新低且 MACD 柱不创新低”来近似实盘更严谨的做法是找局部极值点但回测阶段这样够用。金叉判断用dif dea且上一根dif dea注意上一根 DIF 这里用了近似生产环境应该在缓存里存完整的 dif 序列。参数说明fast/slow/signal是 MACD 标准参数改小会更敏感但假信号多size是每次开仓数量回测时建议固定实盘再按资金比例动态算stop_loss是止损比例0.05 表示亏 5% 平仓这个参数对策略最终收益影响极大后面避坑章节会细说。3. 数据接入与回测引擎把策略挂上去跑出第一条资金曲线3.1 行情数据从哪来、怎么存策略写好了没有数据就是空转。开源交易软件源代码里数据层通常分两种历史数据用于回测实时数据用于实盘。历史数据常见来源是交易所公开的 K 线接口、本地 CSV、或者数据库。我一般会统一成一张表字段固定为datetime, open, high, low, close, volume存成 Parquet 或 SQLite回测时按时间顺序读。import sqlite3 import pandas as pd def load_bars(symbol: str, start: str, end: str, db_pathmarket.db): conn sqlite3.connect(db_path) query SELECT datetime, open, high, low, close, volume FROM bars WHERE symbol ? AND datetime ? AND datetime ? ORDER BY datetime ASC df pd.read_sql_query(query, conn, params(symbol, start, end)) conn.close() # 确保时间列是 datetime 类型回测引擎依赖它排序 df[datetime] pd.to_datetime(df[datetime]) return df.to_dict(records) # 转成 list of dict和 on_bar 的入参对齐这段代码没什么玄学但有两个细节容易翻车。第一ORDER BY datetime ASC必须加否则回测引擎拿到乱序数据资金曲线会莫名其妙。第二返回to_dict(records)而不是 DataFrame是为了让回测引擎和实盘引擎用同一种数据结构喂给策略减少“回测能跑实盘报错”的情况。参数说明symbol是标的代码建议统一大小写start/end用 ISO 格式字符串SQLite 的字符串比较对 ISO 时间有效。如果数据量很大SQLite 会慢可以换 DuckDB 或直接读 Parquet接口不变。3.2 一个 200 行以内的回测引擎核心回测引擎要做的事很纯粹按时间遍历 bar调用策略的on_bar根据返回的指令模拟成交记录资金曲线。下面是一个最小实现能跑单标的、固定手续费和滑点class BacktestEngine: def __init__(self, strategy, bars, initial_cash100000, commission0.0003, slippage0.001): self.strategy strategy self.bars bars self.cash initial_cash self.commission commission # 手续费率万三 self.slippage slippage # 滑点比例千一 self.equity_curve [] # 每根 bar 后的总权益 self.trades [] # 成交记录 def run(self): for bar in self.bars: signal self.strategy.on_bar(bar) action signal[action] size signal[size] if action buy and size 0: price bar[close] * (1 self.slippage) cost price * size * (1 self.commission) if cost self.cash: self.cash - cost self.strategy.update_position(buy, price, size) self.trades.append({datetime: bar[datetime], action: buy, price: price, size: size, reason: signal[reason]}) elif action sell and size 0: price bar[close] * (1 - self.slippage) revenue price * size * (1 - self.commission) self.cash revenue self.strategy.update_position(sell, price, size) self.trades.append({datetime: bar[datetime], action: sell, price: price, size: size, reason: signal[reason]}) # 计算当前总权益现金 持仓市值 position_value self.strategy.position * bar[close] self.equity_curve.append({ datetime: bar[datetime], equity: self.cash position_value }) return self.equity_curve, self.trades逻辑说明买入时价格加上滑点卖出时减去滑点手续费按成交额比例扣。持仓市值用当前 bar 的收盘价估算这是回测的常见近似实盘要用最新价。update_position在成交后调用保证策略内部状态和引擎一致。参数说明commission和slippage是最容易被低估的两个参数。A 股手续费大约万二到万三加上印花税卖出千一实际成本更高滑点取决于流动性和下单方式小盘股千一都算乐观。回测时把这两个参数设得比现实更宽松得到的资金曲线才有参考价值。跑起来只需要几行bars load_bars(BTCUSDT, 2023-01-01, 2024-01-01) strategy MACDDoubleBottom({fast: 12, slow: 26, signal: 9, size: 1, stop_loss: 0.05}) engine BacktestEngine(strategy, bars) equity, trades engine.run() print(f最终权益: {equity[-1][equity]:.2f}, 成交笔数: {len(trades)})到这里一套开源交易软件源代码的最小闭环就通了策略、数据、回测。接下来是实盘对接和避坑。4. 实盘对接与常见坑回测赚钱实盘亏的五个血泪原因4.1 实盘接口的抽象层怎么写实盘和回测最大的区别是回测里成交是瞬间的实盘里下单、成交、撤单都是异步的。所以实盘层不能直接复用回测引擎而要加一个执行适配器。常见做法是定义一个Broker接口回测用SimBroker实盘用具体券商的 API 封装。class Broker(ABC): abstractmethod def send_order(self, symbol, action, size, priceNone): pass abstractmethod def get_position(self, symbol): pass class SimBroker(Broker): # 回测用的模拟券商逻辑和 BacktestEngine 里的成交模拟一致 ... class LiveBroker(Broker): # 实盘券商内部调用具体 API注意处理网络超时和部分成交 ...这样策略代码完全不用改回测和实盘只是换一个 Broker 实例。我一般会在实盘启动时先跑一遍get_position把策略的position和真实持仓对齐否则策略以为空仓、实际有仓第一笔下单就翻车。4.2 避坑五个让回测和实盘分道扬镳的细节现象一回测年化 80%实盘三个月亏 20%。原因回测用了未来函数。比如在on_bar里用了当前 bar 的high/low来判断突破但实盘那一刻 high/low 还没走完。解决策略里只允许用close和之前 bar 的数据需要盘中判断就用更小周期别用当前 bar 的极值。现象二回测成交价总是比实盘好。原因滑点设太小或者用了收盘价成交但实盘是市价单。解决回测买入用close * (1 slippage)卖出用close * (1 - slippage)滑点至少千一如果策略是突破买入实盘往往买在更高滑点要设到千二以上。现象三策略参数在回测里调得很漂亮一换时间段就崩。原因过拟合。参数是在同一段数据上反复调出来的相当于偷看了答案。解决把数据分成训练段和验证段参数只在训练段调验证段跑一次就不许再改。更狠一点用滚动窗口每个窗口重新调参看参数稳定性。现象四实盘偶尔漏单持仓和策略状态不一致。原因网络超时或部分成交后没有回调处理。解决每次下单后记录订单 ID定时用get_position对账策略的position不要自己维护直接从 Broker 查或者用事件驱动的方式在成交回调里更新。现象五回测资金曲线很平滑实盘波动大得睡不着。原因回测没有考虑资金管理和仓位上限每次满仓干。解决在策略层加仓位控制比如单标的最大仓位不超过总资金的 20%总持仓不超过 80%止损不要用固定比例用 ATR 动态算避免波动大的时候被反复扫损。5. 进阶技巧用参数扫描和滚动验证判断策略值不值得上实盘策略写完、回测跑通之后最危险的动作就是直接上实盘。我自己的习惯是先做两件事参数敏感性扫描和滚动前向验证。参数扫描不是找最优参数而是看策略在参数变化时表现是否稳定。如果一个策略在均线周期 10 到 30 之间收益都还行那它大概率不是拟合出来的如果只有周期 17 赚钱、16 和 18 都亏那就是玄学别碰。import itertools def param_scan(strategy_cls, bars, param_grid): results [] keys param_grid.keys() for values in itertools.product(*param_grid.values()): params dict(zip(keys, values)) strategy strategy_cls(params) engine BacktestEngine(strategy, bars) equity, trades engine.run() final equity[-1][equity] results.append({**params, final_equity: final, trades: len(trades)}) return pd.DataFrame(results).sort_values(final_equity, ascendingFalse) # 示例扫描 MACD 参数和止损 grid {fast: [8, 12, 16], slow: [20, 26, 32], stop_loss: [0.03, 0.05, 0.08]} df param_scan(MACDDoubleBottom, bars, grid) print(df.head(10))跑完看两列final_equity的分布以及相邻参数之间的差异。如果前 10 名参数差异很大但收益接近说明策略对参数不敏感是好事如果只有一组参数遥遥领先其他都亏那这组参数大概率是运气。滚动前向验证更接近实盘把数据切成 N 段用第 1 段调参在第 2 段验证再用第 2 段调参在第 3 段验证以此类推。最后把所有验证段拼起来看整体资金曲线。这个曲线才是你上实盘前最该信的曲线。def walk_forward(strategy_cls, bars, param_grid, n_splits5): segment len(bars) // n_splits all_equity [] for i in range(n_splits - 1): train bars[i*segment:(i1)*segment] test bars[(i1)*segment:(i2)*segment] # 在训练段找最优参数 best param_scan(strategy_cls, train, param_grid).iloc[0] params {k: best[k] for k in param_grid.keys()} # 用最优参数在验证段跑 strategy strategy_cls(params) engine BacktestEngine(strategy, test) equity, _ engine.run() all_equity.extend(equity) return all_equity这个做法比单次回测慢很多但能过滤掉大部分自欺欺人的策略。我自己的教训是曾经有一个策略单次回测年化 120%参数扫描发现只有一组参数赚钱滚动验证直接亏 30%。如果当时直接上实盘学费会很贵。所以现在我的习惯是任何策略上实盘前滚动验证的资金曲线必须比单次回测差不太多否则宁可不上。希望帮到你。本文还有配套的精品资源点击获取