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个为NaN |
EMA(X,N) | N周期指数均值 | X.ewm(span=N, adjust=False).mean() | adjust必须为False才和通达信一致 |
REF(X,N) | N周期前的值 | X.shift(N) | N为正表示过去 |
HHV(X,N) | N周期最高值 | X.rolling(N).max() | 含当前bar |
LLV(X,N) | N周期最低值 | X.rolling(N).min() | 含当前bar |
SUM(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数组，可再转Series |
CROSS(A,B) | A上穿B | (A>B) & (A.shift(1)<=B.shift(1)) | 注意边界 |
BARSLAST(COND) | 上次条件成立到现在的周期数 | 需自定义函数，见3.3 | 无直接对应 |
STD(X,N) | N周期标准差 | X.rolling(N).std() | 默认样本标准差，ddof=1 |
ABS(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/(N+1)。pandas的ewm默认adjust=True，会做加权修正，结果和通达信不一样。必须显式设置adjust=False，才能得到完全一致的递推结果。这个坑我见过太多人踩，回测曲线对不上，查半天才发现是这里。
3.2 条件判断与逻辑运算的向量化
通达信里大量使用IF、AND、OR、NOT这些逻辑。在pandas里，布尔序列的运算要用&、|、~，并且每个条件都要用括号包起来，否则运算符优先级会出错。比如：
金叉 := CROSS(MA5, MA10) AND CLOSE > MA20;翻译成Python：
golden_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, index=cond.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 数据准备与标准化
先搞定数据。不管你数据从哪来，最终要整理成这样的DataFrame：
import 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(drop=True) 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);翻译成Python：
def translate_strategy(df): close = df['close'] # 均线 ma5 = close.rolling(5).mean() ma20 = close.rolling(20).mean() # MACD ema12 = close.ewm(span=12, adjust=False).mean() ema26 = close.ewm(span=26, adjust=False).mean() dif = ema12 - ema26 dea = dif.ewm(span=9, adjust=False).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参数对应通达信里的周期N，span=12就是12周期。adjust=False是必须的，前面说过原因。rolling的窗口大小直接对应通达信的N，没有歧义。
CROSS的展开逻辑是(A > B) & (A.shift(1) <= B.shift(1))。这里用<=而不是<，是因为通达信的CROSS定义是“前一根A小于等于B，当前A大于B”。如果你用<，在A和B前一根相等的情况下会漏掉信号。这个细节在实盘中影响不大，但在回测校验时会导致信号数量对不上。
还有一个隐藏参数是min_periods。rolling(N).mean()默认min_periods=N，前N-1个是NaN。如果你希望从第一根就开始计算（用可用数据），可以设min_periods=1，但这和通达信行为不一致。通达信在数据不足时就是返回无效值，所以保持默认即可。
4.4 信号校验与结果比对
写完代码后，一定要做校验。我的做法是：取一段数据，在通达信里导出信号列（可以用DRAWTEXT或者直接看指标线），然后在Python里输出同样的信号，逐行对比。对比时注意两点：一是日期对齐，二是NaN的处理。
如果发现信号数量不一致，按这个顺序排查：
- 检查
EMA是否用了adjust=False。 - 检查
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值对比 | 设adjust=False |
| 前几根K线信号异常 | NaN未处理 | 检查isnull().sum() | fillna(False) |
| 回测速度慢 | 循环内重复计算 | 用cProfile定位 | 预计算所有指标 |
| 复权后价格跳变 | 复权方式不一致 | 对比原始价格 | 统一前复权 |
| 涨跌停信号误判 | 未过滤一字板 | 检查high==low | 加涨跌停过滤 |
这张表里的每一条都是我实际踩过的坑。特别是第一条，当年为了一个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_capital=100000): 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, index=df.index)这个回测逻辑很简单，但足够验证策略的基本表现。注意这里用了循环，因为涉及仓位状态变化，没法完全向量化。不过回测只跑一次，循环的开销可以接受。
6.3 参数优化的正确姿势
参数优化是迁移到Python的核心价值之一。通达信里手动改参数、看曲线，效率极低。Python里可以用网格搜索或者随机搜索，批量跑参数组合。但要注意过拟合问题。我的建议是：先用粗网格找到大致范围，再用细网格局部搜索，最后用样本外数据验证。不要为了追求历史最优参数而过度优化，那样实盘会亏得很惨。
还有一个实用技巧是并行化参数搜索。用joblib或者multiprocessing把不同参数组合分配到多个核心，速度能提升好几倍。我一般会把参数组合生成一个列表，然后用Pool.map并行执行，结果汇总成DataFrame，按夏普比率排序。
6.4 实盘对接的注意事项
如果你打算把迁移后的策略用于实盘，有几个点必须注意。第一，信号确认时间。通达信公式在盘中是实时计算的，但很多指标（比如均线）在收盘前会变化。Python回测用的是收盘价，实盘如果盘中触发，可能会有偏差。建议用收盘后确认信号，次日开盘执行。第二，滑点和手续费。回测里如果不考虑这些，收益率会虚高。一般按万分之三手续费加千分之一滑点估算。第三，数据更新。实盘需要每天更新数据，建议写一个定时任务，收盘后自动拉取数据、计算信号、发送通知。
我个人在实际操作中的体会是，迁移本身不难，难的是保证语义一致和性能达标。只要把函数映射表整理好，把边界条件测试覆盖到，剩下的就是工程问题。真正花时间的往往不是翻译公式，而是数据清洗和结果校验。所以如果你正准备做这件事，建议先把数据管道搭稳，再动手翻译，能省下大量返工的时间。
