Python构建交易系统：技术指标背后的交易认知是核心

做交易系统开发的时候很多人会陷入一个循环不断往系统里加指标、加信号、调参数回测曲线好看几天换一段行情就失效。问题往往不在指标本身而在于指标背后的交易认知没有梳理清楚。本文是交易系统构建系列的第 03 篇围绕“正确交易认知是用好指标做交易的前提”展开用 Python 代码拆解指标模块的设计思路帮助你把对市场的理解沉淀成可运行的规则。1. 为什么交易认知比指标本身更重要1.1 指标只是认知的载体很多人以为“用好指标”等于“找一个胜率高的指标”然后把精力花在参数优化上。实际上指标计算出来的每一个数值都只是对某一段行情特征的量化表达。比如移动平均线代表的是过去一段时间持仓成本的平均水平布林带描述的是价格相对于统计区间的偏离程度MACD 反映的是两段不同周期均线之间的动能变化。指标是数学公式它不会告诉你什么时候该买、什么时候该卖。真正决定买卖动作的是人对市场状态的理解。同样是 RSI 进入 30 以下的超卖区域在震荡行情里可能是反转信号在单边下跌行情里可能是下跌中继。区别不在 RSI 本身而在于交易者是否认清了自己面对的市场状态。1.2 交易认知包含哪些内容在写代码之前我建议先把自己的交易认知拆成几个明确的模块认知维度需要回答的问题对应系统设计市场状态现在是趋势行情还是震荡行情状态识别模块信号定义什么条件触发开仓、平仓信号生成模块风险边界单笔亏损上限是多少止损、仓位控制模块盈亏比预期预期盈利和预期亏损的比例出场策略模块交易频率一天交易几次持仓多久周期级别、过滤规则这些认知如果没有在系统里明确定义指标就只是漂亮的数据图表无法形成稳定的交易决策。1.3 为什么很多指标系统不稳定一个常见的现象是系统在历史数据上表现很好实盘却不行。排除数据泄露和未来函数等问题一个很重要的原因是系统设计时没有把交易认知写进代码。比如只写了“金叉买入”的规则但没有定义什么是“趋势向上”。只计算了 ATR 止损却没有定义市场进入极端行情时如何降低仓位。信号频繁产生系统没有过滤机制导致手续费和滑点吃掉了利润。这些问题对应的不是指标公式有问题而是缺乏一套完整的认知框架。技术指标是“术”交易认知是“道”。道如果乱了术再花哨也没有意义。2. 环境准备与工程结构2.1 开发环境说明本文代码以 Python 为例主要依赖 pandas 和 numpy。版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示设计思路。pip install pandas numpy建议使用 Python 3.8 及以上版本。IDE 可以使用 PyCharm、VS Code也可以直接在 Jupyter Notebook 里运行。为了便于大家理解下面的示例会保持代码尽量精简但结构上会按照实际交易系统的模块化思路组织。2.2 推荐的项目目录结构trading_system/ ├── config/ │ └── strategy_config.yaml ├── data/ │ └── market_data.csv ├── indicators/ │ ├── __init__.py │ ├── moving_average.py │ ├── macd.py │ └── atr.py ├── engine/ │ ├── __init__.py │ ├── data_loader.py │ ├── signal_generator.py │ └── risk_manager.py ├── backtest/ │ ├── __init__.py │ └── backtest_engine.py └── main.py这个结构把“指标计算”和“信号生成”分开核心思想是指标模块只负责计算数值不负责判断买卖。判断逻辑放在信号生成模块这样当你的交易认知发生改变时不需要修改指标公式只需要调整信号规则。3. 把交易认知映射为代码模型3.1 用数据结构表达市场状态交易认知落到代码里的第一步是把市场状态变成可枚举、可计算的对象。我们可以在代码中定义市场状态枚举# 文件路径trading_system/engine/market_state.py from enum import Enum class MarketState(Enum): TRENDING_UP trending_up TRENDING_DOWN trending_down RANGING ranging HIGH_VOLATILITY high_volatility一般来说判断市场状态可以分两步用移动平均线的排列关系判断趋势方向。用 ADX平均趋向指数或价格与均线的距离判断趋势强度。ADX 是一种常用的趋势强度指标它不判断方向只判断趋势的强弱。通常认为 ADX 大于 25 时趋势较强小于 20 时市场更接近震荡。我们可以把 ADX 的计算封装在指标模块中再由信号生成模块调用# 文件路径trading_system/indicators/adx.py import pandas as pd import numpy as np def calculate_adx(df: pd.DataFrame, period: int 14) - pd.Series: high df[high] low df[low] close df[close] up_move high.diff() down_move -low.diff() plus_dm np.where((up_move down_move) (up_move 0), up_move, 0.0) minus_dm np.where((down_move up_move) (down_move 0), down_move, 0.0) tr pd.concat([high - low, (high - close.shift()).abs(), (low - close.shift()).abs()], axis1).max(axis1) atr tr.rolling(windowperiod).mean() plus_di 100 * (pd.Series(plus_dm, indexdf.index).rolling(windowperiod).mean() / atr) minus_di 100 * (pd.Series(minus_dm, indexdf.index).rolling(windowperiod).mean() / atr) dx (plus_di - minus_di).abs() / (plus_di minus_di) * 100 adx dx.rolling(windowperiod).mean() return adx这里的关键点是指标模块只做数学计算返回一个 Series 或 DataFrame不产生任何“买”或“卖”的建议。真正的市场状态判断放在信号生成模块# 文件路径trading_system/engine/market_state.py def judge_market_state(df: pd.DataFrame) - MarketState: adx calculate_adx(df) latest_adx adx.iloc[-1] if latest_adx 25: if df[close].iloc[-1] df[ma20].iloc[-1]: return MarketState.TRENDING_UP else: return MarketState.TRENDING_DOWN else: return MarketState.RANGING信号生成模块根据市场状态决定“要不要使用某个指标信号”。例如在震荡行情里均线金叉的死板信号容易频繁失效此时可以降低信号权重或直接过滤掉。3.2 用仓位管理表达风险认知除了市场状态交易认知里很重要的另一个维度是风险边界。很多人写指标系统时完全不处理仓位只输出 0 和 1 的开仓信号这种系统放到实盘里很容易因为一两次大亏而崩溃。一个比较稳妥的做法是先根据账户总资金确定单笔最大亏损再根据止损距离倒推仓位比例。# 文件路径trading_system/engine/risk_manager.py class RiskManager: def __init__(self, account_balance: float, risk_per_trade: float 0.01): self.account_balance account_balance self.risk_per_trade risk_per_trade def calculate_position_size(self, entry_price: float, stop_loss_price: float) - float: 根据入场价和止损价计算建议仓位数量。 止损距离越大仓位越小止损距离越小仓位越大。 risk_amount self.account_balance * self.risk_per_trade stop_distance abs(entry_price - stop_loss_price) if stop_distance 0: return 0.0 position_size risk_amount / stop_distance return position_size这里把“单笔亏损不超过总资金的 1%”这一条交易认知直接变成了代码逻辑。如果你的交易认知是 2% 或 0.5%只需要修改risk_per_trade参数而不需要修改指标公式。3.3 用信号过滤表达交易耐心另一个常见的认知是“不要什么信号都做”。很多新手写指标系统时恨不得每个 K 线都产生一次交易结果手续费和滑点把利润全部吃掉。在代码层面我们可以给信号生成器增加一个“过滤层”。比如只在 ADX 大于 25 时允许开仓。每天最多只允许开仓一次。开盘后 30 分钟内不执行开仓避免噪音。# 文件路径trading_system/engine/signal_generator.py def generate_signals(df: pd.DataFrame, adx: pd.Series, state: MarketState) - pd.Series: 生成交易信号 1 表示多头开仓 -1 表示空头开仓 0 表示无操作 signals pd.Series(0, indexdf.index) # 只在趋势状态下开仓 if state MarketState.RANGING: return signals # 示例MACD 金叉作为多头信号死叉作为空头信号 macd_line df[macd] signal_line df[signal] golden_cross (macd_line signal_line) (macd_line.shift(1) signal_line.shift(1)) death_cross (macd_line signal_line) (macd_line.shift(1) signal_line.shift(1)) signals[golden_cross] 1 signals[death_cross] -1 # 过滤ADX 小于 25 时忽略信号 signals[adx 25] 0 return signals这就是把“耐心等待趋势确认”这条交易认知写进系统的落地方式。指标公式没有变只是加了一个过滤条件信号质量通常就会有明显提升。4. 完整实战构建一个带认知约束的指标系统下面我们用一个完整的小案例把前面讲到的认知建模、指标计算、信号生成、风险控制串起来。案例的目标是读取一份日线行情数据计算常用技术指标根据市场状态产生交易信号并输出一份简单的回测结果。4.1 准备行情数据为了方便演示我们先生成一份模拟行情数据。实际项目中你可以从数据库、CSV 文件或第三方数据源读取。# 文件路径trading_system/data/generate_data.py import pandas as pd import numpy as np np.random.seed(42) date_rng pd.date_range(start2022-01-01, end2023-12-31, freqD) # 模拟价格走势先上涨、后震荡、再下跌 trend np.concatenate([ np.linspace(100, 150, 200), np.linspace(150, 145, 200), np.linspace(145, 90, 200) np.random.normal(0, 1, 200) ]) close pd.Series(trend, indexdate_rng[:len(trend)]) open_price close.shift(1) np.random.normal(0, 0.5, len(close)) high np.maximum(open_price, close) np.abs(np.random.normal(0, 1, len(close))) low np.minimum(open_price, close) - np.abs(np.random.normal(0, 1, len(close))) df pd.DataFrame({ open: open_price, high: high, low: low, close: close }).dropna() df.to_csv(trading_system/data/market_data.csv) print(df.head())这份数据虽然是人造的但包含了明显的趋势切换和回调波动足够用来演示指标系统的构建过程。4.2 编写指标计算模块在indicators目录下创建technical.py统一封装常用指标计算函数# 文件路径trading_system/indicators/technical.py import pandas as pd import numpy as np def add_ma(df: pd.DataFrame, periods: list) - pd.DataFrame: 添加多条移动平均线 for period in periods: df[fma_{period}] df[close].rolling(windowperiod).mean() return df def add_macd(df: pd.DataFrame, fast: int 12, slow: int 26, signal: int 9) - pd.DataFrame: 添加 MACD 指标 ema_fast df[close].ewm(spanfast, adjustFalse).mean() ema_slow df[close].ewm(spanslow, adjustFalse).mean() df[macd] ema_fast - ema_slow df[signal] df[macd].ewm(spansignal, adjustFalse).mean() df[macd_hist] df[macd] - df[signal] return df def add_atr(df: pd.DataFrame, period: int 14) - pd.DataFrame: 添加 ATR 指标用于衡量波动率 high df[high] low df[low] close df[close] tr pd.concat([ high - low, (high - close.shift()).abs(), (low - close.shift()).abs() ], axis1).max(axis1) df[atr] tr.rolling(windowperiod).mean() return df def add_bollinger_bands(df: pd.DataFrame, period: int 20, num_std: float 2.0) - pd.DataFrame: 添加布林带指标 mid df[close].rolling(windowperiod).mean() std df[close].rolling(windowperiod).std() df[boll_mid] mid df[boll_upper] mid num_std * std df[boll_lower] mid - num_std * std return df每个函数只做一件事且不直接产生交易信号。这样做的优点是方便单元测试也方便以后替换指标公式实现。4.3 编写信号生成模块信号生成模块是交易认知的核心落地位置。这里我们定义一个简单的认知规则只做趋势行情震荡行情不开仓。趋势向上的多头信号收盘价上穿 20 日均线且 MACD 在零轴上方。多头平仓信号收盘价下穿 10 日均线。单笔风险控制在账户余额的 1%。# 文件路径trading_system/engine/signal_generator.py import pandas as pd def generate_signals(df: pd.DataFrame) - pd.DataFrame: 根据交易认知生成交易信号。 返回的 DataFrame 中包含 position 列1 表示持仓0 表示空仓。 signals pd.DataFrame(indexdf.index) signals[close] df[close] signals[ma_10] df[ma_10] signals[ma_20] df[ma_20] signals[macd] df[macd] signals[adx] df[adx] # 初始状态空仓 signals[signal] 0 positions [] in_position False for i in range(len(signals)): row signals.iloc[i] close row[close] ma_10 row[ma_10] ma_20 row[ma_20] macd row[macd] adx row[adx] # 趋势过滤ADX 大于 25 才允许开仓 if not in_position: if adx 25 and macd 0 and close ma_20: signals.loc[signals.index[i], signal] 1 in_position True else: signals.loc[signals.index[i], signal] 0 else: # 持仓中收盘价跌破 10 日均线则平仓 if close ma_10: signals.loc[signals.index[i], signal] -1 in_position False else: signals.loc[signals.index[i], signal] 0 # 将信号转换为持仓状态 signals[position] signals[signal].replace(0, pd.NA).ffill().fillna(0) return signals这里要注意信号生成模块使用了循环来跟踪持仓状态。虽然循环性能不如向量化操作但胜在逻辑清晰适合表达带状态的交易规则。实际项目如果数据量很大可以改用 numba 或 Cython 优化。4.4 编写回测模块回测模块负责把信号和行情数据结合起来计算每日权益曲线和最终收益。这里为了保持简单只计算固定数量的持仓收益不处理手续费和滑点但会在结果中给出手续费影响提示。# 文件路径trading_system/backtest/backtest_engine.py import pandas as pd def run_backtest(signals: pd.DataFrame, initial_balance: float 100000.0) - pd.DataFrame: 简单的回测引擎 不考虑手续费和滑点时收益等于持仓状态乘以下一日收益率。 df signals.copy() df[daily_return] df[close].pct_change().fillna(0) # 信号信号在当天收盘时确定因此收益从下一日开始计算 df[strategy_return] df[position].shift(1) * df[daily_return] df[strategy_return] df[strategy_return].fillna(0) df[balance] initial_balance * (1 df[strategy_return]).cumprod() df[benchmark_balance] initial_balance * (1 df[daily_return]).cumprod() return df运行回测后打印关键指标# 文件路径trading_system/main.py from data.generate_data import get_data from indicators.technical import add_ma, add_macd, add_atr, add_bollinger_bands from indicators.adx import calculate_adx from engine.signal_generator import generate_signals from backtest.backtest_engine import run_backtest # 1. 读取数据 df get_data() # 2. 计算指标 df add_ma(df, periods[10, 20]) df add_macd(df) df add_atr(df) df add_bollinger_bands(df) df[adx] calculate_adx(df) # 3. 删除前期的 NA 数据 df df.dropna().copy() # 4. 生成信号 signals generate_signals(df) # 5. 回测 result run_backtest(signals) # 6. 输出统计 total_return result[balance].iloc[-1] / 100000 - 1 benchmark_return result[benchmark_balance].iloc[-1] / 100000 - 1 days len(result) annual_return (1 total_return) ** (252 / days) - 1 print(f策略总收益率: {total_return:.2%}) print(f基准总收益率: {benchmark_return:.2%}) print(f年化收益率: {annual_return:.2%}) print(f最终权益: {result[balance].iloc[-1]:.2f})预期输出不会特别惊艳因为这份模拟数据本身存在明显趋势切换而我们的规则并不完美。这个案例的核心不是追求高收益而是演示“认知 → 规则 → 代码 → 回测”的完整链路。4.5 结果说明从回测结果中你通常会看到几个现象趋势行情阶段策略净值跟随上涨。震荡行情阶段策略因为 ADX 过滤而减少交易净值保持平稳。下跌行情阶段策略因为“收盘价跌破 10 日均线”离场避免了大幅回撤。这些现象说明交易认知对系统的约束正在发挥作用。如果去掉 ADX 过滤震荡阶段会出现大量假信号如果去掉均线离场规则下跌阶段会抗单很久。5. 常见问题与排查思路5.1 信号过多现象回测日志里交易次数非常多几乎每天都有开平仓。原因信号过滤条件太少或者指标参数周期太短。排查步骤打印信号出现的日期观察是否集中在震荡区间。检查过滤条件是否真正生效比如 ADX 是否参与了判断。观察手续费和滑点成本占比。解决办法增加趋势过滤条件。拉长指标周期。增加连续确认机制比如连续两根 K 线满足条件才开仓。5.2 回测结果与实盘差异巨大现象历史回测收益很高实盘交易却不断亏损。原因可能存在前视偏差、未来函数或者没有考虑滑点和手续费。排查步骤检查信号生成是否使用了当日收盘后才确定的数据。确认没有使用未来数据进行填充。加入手续费和滑点后重新回测。这里特别提醒回测中常见的“未来函数”包括使用了当日最高价、最低价产生信号然后在当日收盘价成交实际中这是无法做到的。5.3 极端行情下止损被突破现象设置了 ATR 止损但行情跳空巨大止损价直接穿过。原因止损指令在极端行情里可能无法按预期价格成交。解决办法在仓位计算中保留一定缓冲。结合波动率调整止损距离。在系统里加入最大回撤熔断机制当账户回撤超过阈值时暂停交易。问题现象常见原因解决思路信号过多过滤条件不足增加趋势和波动过滤回测收益虚高未来函数、前视偏差检查数据对齐加入手续费实盘滑点大流动性不足减少交易频率限价单止损频繁被扫止损距离太小用 ATR 动态止损参数过拟合手动反复调参使用样本外验证6. 最佳实践与工程建议6.1 指标模块保持纯净指标计算函数不应该修改外部状态也不应该依赖全局配置。每次传入 DataFrame返回新的 DataFrame 或 Series。这样方便单元测试也方便以后接入实时行情时复用同一个函数。6.2 交易认知用配置沉淀不要硬编码认知规则。可以把“单笔风险比例”“ADX 阈值”“是否允许震荡开仓”这些参数写入 YAML 配置文件# 文件路径trading_system/config/strategy_config.yaml risk: risk_per_trade: 0.01 max_drawdown: 0.20 filter: adx_threshold: 25 allow_ranging: false signal: ma_short: 10 ma_long: 20 use_macd_zero_filter: true然后在代码中读取配置import yaml with open(trading_system/config/strategy_config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) risk_per_trade config[risk][risk_per_trade] adx_threshold config[filter][adx_threshold]这样做的好处是当你的交易认知进化时只需要调整配置不需要改动代码逻辑。6.3 警惕参数过拟合很多交易系统死在“参数优化”上。历史数据里把 MA 从 20 改成 21回测收益提高 5%于是就觉得找到了圣杯。实际上这种优化很可能是在拟合历史噪音。我的建议是设置训练集和验证集不要在同一段数据上反复调参。参数变化对结果影响过大时说明系统本身不稳定。优先减少参数数量而不是增加参数数量。用更长的历史数据和不同市场状态的数据验证。6.4 日志与可追踪性交易系统要能回答“这笔交易为什么开仓”“为什么平仓”。建议在信号生成时记录完整的决策上下文import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(message)s) def log_decision(date, action, adx, macd, close, ma_20): logging.info( f{date} {action}: close{close:.2f}, ma_20{ma_20:.2f}, fmacd{macd:.2f}, adx{adx:.2f} )这样可以回放任意一笔交易找到亏损的根源。6.5 生产环境的安全边界交易系统涉及真金白银代码质量和风控纪律更重要权限上要最小化回测、实盘、管理端分开账号。实盘接口要加幂等保护避免重复下单。数据库操作用事务对订单表修改要先备份。建议先做模拟盘验证一段时间再切换小额实盘。所有参数变更走配置审批流程不要线上随意改。7. 总结与学习路线这一篇的核心不是教你写一个万能指标而是帮你建立一套“交易认知驱动代码”的思维方式。技术指标只是工具真正决定系统稳定性的是你对市场状态、风险边界、信号过滤的理解深度。下一步你可以继续学习如何设计完整的回测框架。如何接入实时行情数据。如何把策略从日线级别扩展到分钟级别。如何用机器学习辅助判断市场状态。如何做更严谨的参数寻优和样本外验证。如果你正在构建自己的交易系统建议先别急着堆指标。花时间把交易认知整理成几条明确的规则再动手写代码这样系统的生命力和可维护性会好很多。如果本文对你有帮助可以收藏备用如果你有更好的指标模块设计思路也欢迎在评论区交流。