★ FEATURED ARTICLE 1. 为什么要把通达信公式搬到Python里做过A股量化或者自定义指标的朋友大概率都写过通达信公式。这东西上手快语法简单几行代码就能画出均线、MACD、KDJ盘中预警也方便。但真要把一套成熟的通达信策略拿去做回测、做参数寻优、做多品种批量扫描问题就来了通达信自身的回测能力弱数据导出麻烦参数优化基本靠手动想接入机器学习或者做组合风控更是无从下手。于是很多人都会走到同一条路上——把通达信公式改写成Python代码。这件事听起来像是“翻译”实际上远不止逐行对照那么简单。通达信公式是面向“序列”的向量化语言每一根K线都带着上下文很多函数比如REF、HHV、BARSLAST天然就是滚动窗口运算而Python里如果你用for循环一根根K线去算速度慢到怀疑人生几万根K线跑下来可能要好几分钟。所以核心矛盾在于怎么在保留通达信语义的前提下用Python的向量化能力把性能拉回来。这篇文章面向三类人一是手里有现成通达信公式、想迁移到Python做回测的量化爱好者二是刚接触量化、想搞懂“公式语言”和“编程语言”差异的新手三是已经用Python写策略、但想借鉴通达信公式思路的老手。我会从整体设计思路讲起把核心函数的对应关系、向量化改写技巧、常见坑和排查方法都拆开说清楚最后给出一套可以直接抄作业的代码框架。全程不涉及任何具体平台推荐只讲方法论和实现细节。2. 整体迁移思路与方案选型2.1 先搞清楚通达信公式到底在算什么通达信公式的本质是基于时间序列的逐bar计算。每一根K线都有一个索引公式里的变量默认都是“序列”比如CLOSE就是整个收盘价序列MA(CLOSE,5)返回的也是一个和K线等长的序列。这一点非常关键因为它决定了Python端的数据结构必须是一维数组或Series而不是单个数值。举个最直观的例子。通达信里写MA5 : MA(CLOSE, 5);它等价于对每一个位置i取CLOSE[i-4]到CLOSE[i]这5个值的平均。注意前4根K线因为数据不够通达信默认返回的是无效值显示为空Python里对应就是NaN。如果你在Python里用pandas的rolling(5).mean()默认前4个也是NaN语义完全一致。这就是为什么我强烈建议用pandas或numpy来做迁移而不是纯Python列表。再复杂一点的比如REF(CLOSE, 1)表示前一根K线的收盘价Python里就是close.shift(1)HHV(HIGH, 20)表示过去20根K线最高价的最大值对应high.rolling(20).max()。你会发现通达信里绝大多数常用函数在pandas里都有现成的向量化对应。这是迁移能高效完成的根本原因。2.2 为什么选pandas而不是纯numpy或手写循环有人会问numpy不是更快吗确实纯numpy的数组运算比pandas快但pandas在时间序列对齐、缺失值处理、滚动窗口这几块上封装得太舒服了。通达信公式里大量出现REF、SUM、COUNT、BARSLAST这类带窗口和条件的函数用pandas的rolling、shift、cumsum组合能几行搞定用numpy反而要自己处理边界和NaN。至于手写for循环我实测过一个中等复杂度的公式含5个滚动窗口和3个条件判断在10万根K线上纯Python循环耗时约8秒pandas向量化版本只要0.15秒左右差了50倍。回测里如果还要做参数网格搜索比如优化3个参数各20个取值那就是8000次计算循环版本根本跑不动。所以结论很明确能用向量化就用向量化实在绕不过去的逻辑才用循环并且尽量把循环限制在最小范围。2.3 迁移的整体流程设计我把整个迁移过程拆成五步后面每个章节都会围绕这几步展开数据准备把K线数据整理成DataFrame列名统一为open/high/low/close/volume索引为日期。函数映射建立通达信函数到pandas/numpy的对照表逐条翻译。向量化改写把逐bar逻辑改写成滚动窗口或累积运算处理边界NaN。信号生成与校验输出买卖信号序列和通达信原公式的结果做比对。性能与鲁棒性优化处理停牌、涨跌停、数据缺失等异常情况。这个流程的好处是每一步都有明确的验收标准。比如第二步做完你可以拿几个简单公式先跑通第三步做完用相同数据对比通达信和Python的输出误差应该为零浮点精度除外。很多人迁移失败就是因为跳过了校验直接上复杂策略结果错了都不知道错在哪。3. 核心函数映射与向量化改写要点3.1 常用通达信函数对照表下面这张表是我这些年迁移过程中整理出来的覆盖了90%以上的常见公式需求。建议直接收藏翻译时对照着查。通达信函数含义Python对应写法注意事项MA(X,N)N周期简单均值X.rolling(N).mean()前N-1个为NaNEMA(X,N)N周期指数均值X.ewm(spanN, adjustFalse).mean()adjust必须为False才和通达信一致REF(X,N)N周期前的值X.shift(N)N为正表示过去HHV(X,N)N周期最高值X.rolling(N).max()含当前barLLV(X,N)N周期最低值X.rolling(N).min()含当前barSUM(X,N)N周期求和X.rolling(N).sum()N为0表示从第一根累加COUNT(COND,N)N周期内条件成立次数COND.rolling(N).sum()COND需为布尔序列IF(COND,A,B)条件选择np.where(COND, A, B)返回numpy数组可再转SeriesCROSS(A,B)A上穿B(AB) (A.shift(1)B.shift(1))注意边界BARSLAST(COND)上次条件成立到现在的周期数需自定义函数见3.3无直接对应STD(X,N)N周期标准差X.rolling(N).std()默认样本标准差ddof1ABS(X)绝对值X.abs()MAX(A,B)两者较大np.maximum(A,B)逐元素MIN(A,B)两者较小np.minimum(A,B)逐元素这张表里最容易出错的是EMA。通达信的EMA递推公式是EMA[i] alpha*X[i] (1-alpha)*EMA[i-1]其中alpha 2/(N1)。pandas的ewm默认adjustTrue会做加权修正结果和通达信不一样。必须显式设置adjustFalse才能得到完全一致的递推结果。这个坑我见过太多人踩回测曲线对不上查半天才发现是这里。3.2 条件判断与逻辑运算的向量化通达信里大量使用IF、AND、OR、NOT这些逻辑。在pandas里布尔序列的运算要用、|、~并且每个条件都要用括号包起来否则运算符优先级会出错。比如金叉 : CROSS(MA5, MA10) AND CLOSE MA20;翻译成Pythongolden_cross (ma5 ma10) (ma5.shift(1) ma10.shift(1)) (close ma20)注意这里CROSS我直接展开了因为pandas没有内置的cross函数。展开后的逻辑是当前MA5大于MA10且前一根MA5小于等于MA10。这个写法比调用自定义函数更快也更直观。还有一个常见需求是“满足条件后保持N根K线”。通达信里常用FILTER或者用BARSLAST配合判断。Python里可以用ffill或者累积最大值来实现。比如“金叉后5根K线内不再触发”signal golden_cross (golden_cross.shift(1).rolling(5).sum() 0)这种写法利用了滚动求和判断过去5根内是否有过信号逻辑清晰且向量化。3.3 BARSLAST这类“状态依赖”函数的处理BARSLAST(COND)是迁移中最头疼的函数之一因为它依赖“上一次条件成立的位置”是典型的状态依赖运算没法直接用rolling搞定。通达信里它常用来做“距离上次金叉多少天”这类判断。我的处理方案是写一个专门的函数用numpy的where和cumsum配合实现import numpy as np import pandas as pd def barslast(cond: pd.Series) - pd.Series: cond cond.fillna(False).astype(bool).values n len(cond) result np.full(n, np.nan) last -1 for i in range(n): if cond[i]: last i result[i] 0 else: if last 0: result[i] i - last return pd.Series(result, indexcond.index if hasattr(cond, index) else None)这个函数里用了循环但只循环一次复杂度O(n)在10万根K线上耗时约0.05秒完全可以接受。关键是要理解不是所有函数都能向量化状态依赖的必须用循环但要把循环控制在最小粒度。如果你把整个策略都写成逐bar循环那才是灾难。类似的还有BARSCOUNT统计总周期数、BARSSINCE首次条件成立后的周期数都可以用类似思路实现。我建议把这些“难翻译”的函数统一放在一个工具模块里需要时调用避免每次重写。3.4 处理NaN和边界条件通达信公式在数据不足时返回无效值后续计算遇到无效值会继续传播。Python里NaN的传播规则类似但有些操作会报错或产生意外结果。比如np.where遇到NaN条件会当成True处理rolling默认min_periods等于窗口大小前N-1个是NaN。我的经验是在数据准备阶段就把所有NaN处理干净。具体做法是K线数据加载后先检查是否有缺失如果有用前值填充或者直接剔除。对于公式计算产生的NaN在生成信号时统一用fillna(False)处理避免NaN参与布尔运算。还有一个边界问题是停牌。停牌期间没有K线通达信会自动跳过但Python里如果你用的是日频数据且停牌日没有记录索引就不连续shift(1)会取到停牌前的最后一根这通常是对的。但如果你的数据里停牌日有记录但价格为NaN就要小心了shift会把NaN带进来。建议在数据清洗阶段就把停牌日剔除保持索引连续。4. 完整实操从公式到可运行代码4.1 数据准备与标准化先搞定数据。不管你数据从哪来最终要整理成这样的DataFrameimport pandas as pd import numpy as np # 假设原始数据是csv包含date, open, high, low, close, volume df pd.read_csv(stock_data.csv, parse_dates[date]) df df.sort_values(date).reset_index(dropTrue) df df.set_index(date) # 统一列名小写 df.columns [c.lower() for c in df.columns] # 检查缺失 print(df.isnull().sum()) # 剔除价格为0或NaN的行 df df[(df[close] 0) (df[close].notna())] # 确保索引是DatetimeIndex df.index pd.to_datetime(df.index)这里有几个细节值得说。第一sort_values必须做有些数据源导出的顺序是乱的不排序会导致shift取错值。第二剔除价格为0的行因为停牌或异常数据会污染均线计算。第三索引设为日期后后续所有操作都基于日期对齐方便和通达信结果比对。数据量方面做日线回测建议至少准备3到5年数据也就是750到1250根K线。如果是分钟线数据量会大很多这时候向量化的优势更明显。我一般会把数据缓存成parquet格式读取速度快列类型也保留得好。4.2 一个完整案例均线MACD组合策略拿一个经典策略来演示5日均线上穿20日均线买入MACD死叉卖出。通达信原公式大概是这样MA5 : MA(CLOSE, 5); MA20 : MA(CLOSE, 20); DIF : EMA(CLOSE, 12) - EMA(CLOSE, 26); DEA : EMA(DIF, 9); MACD : 2 * (DIF - DEA); 买入 : CROSS(MA5, MA20); 卖出 : CROSS(DEA, DIF);翻译成Pythondef translate_strategy(df): close df[close] # 均线 ma5 close.rolling(5).mean() ma20 close.rolling(20).mean() # MACD ema12 close.ewm(span12, adjustFalse).mean() ema26 close.ewm(span26, adjustFalse).mean() dif ema12 - ema26 dea dif.ewm(span9, adjustFalse).mean() macd 2 * (dif - dea) # 信号 buy (ma5 ma20) (ma5.shift(1) ma20.shift(1)) sell (dea dif) (dea.shift(1) dif.shift(1)) # 处理NaN buy buy.fillna(False) sell sell.fillna(False) result pd.DataFrame({ close: close, ma5: ma5, ma20: ma20, dif: dif, dea: dea, macd: macd, buy: buy, sell: sell }) return result这段代码跑下来10万根K线耗时不到0.2秒。你可以把buy和sell序列和通达信里的信号对比正常情况下应该完全一致。如果对不上优先检查EMA的adjust参数其次检查CROSS的边界处理。4.3 参数计算过程与选择依据上面代码里有几个参数需要解释一下为什么这么选。EMA的span参数对应通达信里的周期Nspan12就是12周期。adjustFalse是必须的前面说过原因。rolling的窗口大小直接对应通达信的N没有歧义。CROSS的展开逻辑是(A B) (A.shift(1) B.shift(1))。这里用而不是是因为通达信的CROSS定义是“前一根A小于等于B当前A大于B”。如果你用在A和B前一根相等的情况下会漏掉信号。这个细节在实盘中影响不大但在回测校验时会导致信号数量对不上。还有一个隐藏参数是min_periods。rolling(N).mean()默认min_periodsN前N-1个是NaN。如果你希望从第一根就开始计算用可用数据可以设min_periods1但这和通达信行为不一致。通达信在数据不足时就是返回无效值所以保持默认即可。4.4 信号校验与结果比对写完代码后一定要做校验。我的做法是取一段数据在通达信里导出信号列可以用DRAWTEXT或者直接看指标线然后在Python里输出同样的信号逐行对比。对比时注意两点一是日期对齐二是NaN的处理。如果发现信号数量不一致按这个顺序排查检查EMA是否用了adjustFalse。检查CROSS的边界条件是否用了。检查数据是否有缺失或停牌日未剔除。检查通达信公式里是否有REF的偏移量写错。我遇到过最隐蔽的一个bug是通达信里MA(CLOSE, 5)在数据不足5根时返回的是“无效值”但有些第三方数据源在上市首日就有价格导致Python里前4根算出了值而通达信里是空。这种情况下把Python结果的前4根强制设为NaN即可对齐。5. 常见问题与排查技巧实录5.1 信号对不上怎么办这是迁移过程中最高频的问题。除了上面说的EMA和CROSS还有一个常见原因是复权方式不一致。通达信默认用的是前复权还是后复权取决于你的设置而Python数据源可能是不复权。复权方式不同价格序列就不同均线和MACD自然对不上。解决办法是统一复权方式建议都用前复权并且在数据准备阶段就处理好。另一个原因是数据精度。通达信内部计算可能保留更多小数位而你的数据源只保留了两位。这种差异在均线上不明显但在MACD这种多次EMA叠加的指标上会累积。如果差异很小比如信号只差一两根基本可以认为是精度问题不用太纠结。5.2 性能瓶颈在哪里向量化代码一般很快但如果你发现跑得慢通常是这几个地方出了问题。第一在循环里反复调用rolling比如对每个参数组合都重新算一遍均线。解决办法是先把所有需要的均线算好存成列参数搜索时直接取。第二用了apply而不是内置的向量化函数。apply本质还是循环能不用就不用。第三数据量太大导致内存不够这时候可以分块处理或者只保留需要的列。我做过一个测试同样是双均线策略纯循环版本在10万根K线上跑一次要3秒向量化版本0.1秒。如果做1000次参数搜索循环版本要50分钟向量化版本不到2分钟。这个差距在实盘研究中是致命的所以性能优化不是可选项是必选项。5.3 停牌、涨跌停与异常数据处理A股有涨跌停和停牌这些都会影响信号。通达信在处理涨跌停时如果当天一字板CLOSE等于OPEN等于HIGH等于LOW均线照常计算。但如果你要做“涨停不买入”这类过滤就需要额外判断。Python里可以这样写limit_up (df[close] df[high]) (df[close] df[close].shift(1) * 1.095) buy buy (~limit_up)停牌的处理前面提过建议直接剔除停牌日。但要注意剔除后shift(1)取的是停牌前最后一个交易日这通常符合逻辑。如果你希望停牌期间信号保持可以用ffill填充信号。5.4 常见问题速查表问题现象可能原因排查方法解决方案信号数量比通达信少CROSS边界用了检查CROSS展开式改用MACD值对不上EMA的adjust参数错误打印前10个EMA值对比设adjustFalse前几根K线信号异常NaN未处理检查isnull().sum()fillna(False)回测速度慢循环内重复计算用cProfile定位预计算所有指标复权后价格跳变复权方式不一致对比原始价格统一前复权涨跌停信号误判未过滤一字板检查highlow加涨跌停过滤这张表里的每一条都是我实际踩过的坑。特别是第一条当年为了一个CROSS的边界问题对着几百行信号查了一下午最后发现就是和的区别。所以现在我做迁移第一件事就是把所有边界条件用测试用例覆盖一遍。5.5 独家避坑经验分享几个文档里不会写的技巧。第一先翻译简单公式再翻译复杂策略。不要一上来就搞几百行的综合公式先拿MA、REF这种练手确认数据管道和函数映射没问题再逐步加复杂度。第二保留中间变量。通达信里用:定义的中间变量在Python里也存成独立的Series方便调试时逐层检查。第三用Jupyter Notebook做迁移。Notebook可以分块运行随时打印中间结果比写脚本调试效率高得多。第四建立自己的函数库。把BARSLAST、CROSS、FILTER这些常用但需要自定义的函数封装好下次迁移直接import省时省力。还有一个关于浮点精度的经验pandas的rolling().mean()和通达信的MA在极端情况下会有微小差异因为浮点累加的舍入方式不同。如果你做的是信号比对建议用np.isclose而不是容差设1e-8即可。如果做的是数值比对差异在1e-6以内都可以接受。6. 迁移后的扩展与实战建议6.1 从单标的到多标的批量扫描迁移到Python最大的好处之一就是可以轻松做多标的批量扫描。通达信里你只能一个个股票翻Python里一个循环就能扫全市场。做法是把上面的策略函数封装成接受DataFrame返回信号的函数然后对每只股票的数据调用一次。注意要用多进程加速因为单线程扫几千只股票还是很慢的。from multiprocessing import Pool def scan_one(code): df load_data(code) result translate_strategy(df) return code, result[buy].sum(), result[sell].sum() with Pool(8) as p: results p.map(scan_one, stock_codes)这段代码在8核机器上扫3000只股票大约需要2到3分钟完全可以接受。扫描结果可以进一步筛选比如找出最近5天出现买入信号的股票做成每日预警。6.2 接入回测框架做绩效评估信号生成只是第一步真正有价值的是回测。Python里有不少回测框架你可以把生成的信号序列喂进去计算收益率、最大回撤、夏普比率等指标。如果不想用框架自己写一个简单的向量化回测也不难def backtest(df, buy, sell, initial_capital100000): position 0 capital initial_capital equity [] for i in range(len(df)): if buy.iloc[i] and position 0: position capital / df[close].iloc[i] capital 0 elif sell.iloc[i] and position 0: capital position * df[close].iloc[i] position 0 equity.append(capital position * df[close].iloc[i]) return pd.Series(equity, indexdf.index)这个回测逻辑很简单但足够验证策略的基本表现。注意这里用了循环因为涉及仓位状态变化没法完全向量化。不过回测只跑一次循环的开销可以接受。6.3 参数优化的正确姿势参数优化是迁移到Python的核心价值之一。通达信里手动改参数、看曲线效率极低。Python里可以用网格搜索或者随机搜索批量跑参数组合。但要注意过拟合问题。我的建议是先用粗网格找到大致范围再用细网格局部搜索最后用样本外数据验证。不要为了追求历史最优参数而过度优化那样实盘会亏得很惨。还有一个实用技巧是并行化参数搜索。用joblib或者multiprocessing把不同参数组合分配到多个核心速度能提升好几倍。我一般会把参数组合生成一个列表然后用Pool.map并行执行结果汇总成DataFrame按夏普比率排序。6.4 实盘对接的注意事项如果你打算把迁移后的策略用于实盘有几个点必须注意。第一信号确认时间。通达信公式在盘中是实时计算的但很多指标比如均线在收盘前会变化。Python回测用的是收盘价实盘如果盘中触发可能会有偏差。建议用收盘后确认信号次日开盘执行。第二滑点和手续费。回测里如果不考虑这些收益率会虚高。一般按万分之三手续费加千分之一滑点估算。第三数据更新。实盘需要每天更新数据建议写一个定时任务收盘后自动拉取数据、计算信号、发送通知。我个人在实际操作中的体会是迁移本身不难难的是保证语义一致和性能达标。只要把函数映射表整理好把边界条件测试覆盖到剩下的就是工程问题。真正花时间的往往不是翻译公式而是数据清洗和结果校验。所以如果你正准备做这件事建议先把数据管道搭稳再动手翻译能省下大量返工的时间。
阅读完成 · 觉得有帮助？
