最近在和一些做量化交易的朋友聊天发现一个挺有意思的现象很多人花了大把时间研究策略、调参、回测但真正把策略从“能跑”变成“能稳定、可维护地跑”这一步却总是磕磕绊绊。策略代码、数据源、回测引擎、风控模块、日志系统……这些组件像一堆散落的乐高积木每次想换个策略或者加个新数据源都得重新“搭”一遍过程繁琐不说还容易出错。这让我想起了软件工程里一个老生常谈的问题如何把一次性的、临时的脚本变成一套可复用、可迭代、可协作的工程化系统。在量化领域这个问题尤其突出。你可能会用backtrader、zipline或者自己写一套回测但如何管理不同策略的版本如何确保生产环境和回测环境一致如何优雅地处理实时数据流和批量历史数据如何把策略部署到实盘时还能方便地监控和干预这时候一个清晰的工程化框架就显得至关重要。它不直接帮你赚钱但它能让你赚钱的工具变得更可靠、更高效。今天我们不谈具体的阿尔法策略而是聚焦于如何“搭建”这个工具本身。我们将围绕Harness Engineering这个理念手把手构建一个属于你自己的、可扩展的量化框架。这里的“Harness”不是某个特定软件而是一种“约束与引导”的工程思想——通过一套约定、工具和流程把散乱的量化开发活动“套”进一个高效、规范的轨道里。1. 为什么你的量化策略需要一个“工程化框架”在深入动手之前我们必须先达成一个共识为什么不能直接写个脚本就跑工程化框架到底解决了什么核心痛点很多人对框架有抵触觉得是“过度设计”增加了学习成本。但事实上缺乏框架的“自由”在项目稍微复杂一点后会带来更大的混乱成本。1.1 从“一次性实验”到“可复用资产”的转变量化策略开发天然具有实验性质。你可能会尝试几十个想法最终只有一两个能进入实盘。如果没有框架每个实验可能都是一个独立的、结构各异的脚本。时间一长你会发现代码无法复用A策略的数据处理逻辑写得很好但B策略无法直接调用因为数据加载、清洗的代码和策略逻辑紧耦合在一起。结果无法复现三个月后你发现某个曾经有效的策略失效了想回头检查当时的回测结果和参数却发现找不到当时的代码版本、数据快照和环境配置。协作成为噩梦当你需要和同伴一起开发时光是统一代码风格、定义数据接口、约定回测流程就要耗费大量沟通成本。一个工程化框架的核心价值就是帮你把每一次实验的“副产品”——比如数据获取模块、回测引擎接口、绩效分析工具——沉淀下来变成团队共享的、标准化的“资产”。下一次新策略开发你只需要关心策略逻辑本身其他“轮子”都是现成的、经过测试的。1.2 管理复杂性策略、数据、执行的三重奏一个完整的量化系统至少包含三个核心维度策略产生交易信号的逻辑。它可能是基于技术指标、基本面数据、机器学习模型甚至是另类数据。数据策略的“燃料”。包括历史数据用于回测和实时数据用于实盘数据源可能来自数据库、CSV文件、API接口格式和频率各异。执行如何将策略信号转化为实际的交易。这包括回测时的模拟执行以及实盘时的真实订单执行。如果没有框架这三者通常会混杂在一起。一个脚本里可能既有关闭数据库连接又有计算移动平均线还有打印交易日志。这种混杂使得任何一部分的修改都风险极高。框架的作用就是强制性地进行关注点分离通过清晰的接口和模块边界让策略开发者、数据工程师、运维人员可以相对独立地工作。1.3 Harness Engineering不是工具而是“约束的艺术”“Harness”这个词原意是马具引申为“利用并控制”。Harness Engineering 的精髓在于通过设计一套合理的约束框架、规范、工具链来释放开发者的生产力并保障系统的可靠性。它不是一个现成的软件虽然有很多工具可以辅助而是一种设计思想。对于量化框架来说这意味着约束项目结构强制规定代码、配置、数据、日志的存放位置。所有人都按同一套模板开始新人上手极快。约束数据流定义清晰的数据接口Input/Output所有模块都通过标准接口交换数据避免隐式依赖。约束执行流程将回测、优化、实盘部署定义为一个个可配置的“任务”或“流水线”使其可重复、可监控。约束配置管理将策略参数、数据库连接信息、API密钥等与环境相关的配置集中管理并与代码分离便于不同环境开发、测试、生产的切换。这种“约束”看似限制了自由实则提供了更大的自由——在框架划定的安全边界内你可以专注于创造性的策略研究而不用为工程细节分心。2. 搭建量化框架的核心模块与设计原则理解了“为什么”我们来看“是什么”。一个典型的、基于 Harness Engineering 思想的量化框架应该包含哪些核心模块它们之间如何协作这里我们抽象出一个通用的分层架构。2.1 分层架构从数据到决策的清晰流水线一个健壮的框架通常采用分层设计每一层职责单一上层依赖下层提供的服务。一个经典的量化框架可以划分为以下四层层级核心职责关键模块举例设计原则数据层获取、存储、提供标准化数据数据获取器、数据清洗器、数据存储器、数据查询接口接口标准化为上层提供统一的数据访问接口如get_bars(symbol, start_date, end_date, frequency)隐藏底层数据源数据库、CSV、API的差异。可插拔方便更换或增加新的数据源。策略层生成交易信号策略基类、技术指标库、信号计算引擎纯逻辑策略类只负责接收数据、计算信号不关心数据从哪里来、信号发到哪里去。可回测策略逻辑必须能在历史数据上独立运行用于回测。执行层模拟或真实执行交易管理投资组合状态回测引擎、实盘交易网关、投资组合管理器、风险控制器模拟与实盘统一回测引擎和实盘交易网关应实现相同的执行接口使得策略无需修改即可在两种模式下运行。状态管理精确记录现金、持仓、交易记录、绩效等状态。应用层组织工作流提供用户界面回测任务运行器、参数优化器、可视化模块、监控告警模块任务化将一次完整的回测或实盘运行定义为一个可配置、可重复执行的任务。可观测提供丰富的日志、指标和可视化结果便于分析和调试。这个分层架构是 Harness Engineering 思想的具体体现。每一层都通过明确的接口为上层提供服务同时受到下层接口的约束。修改数据源不会影响策略逻辑更换券商API也只需改动执行层的网关模块。2.2 核心模块详解与代码结构让我们把这些抽象的概念落地到一个具体的项目目录结构中。以下是一个建议的框架项目布局quant_framework/ ├── config/ # 配置管理 │ ├── default.yaml # 默认配置 │ ├── development.yaml # 开发环境配置 │ └── production.yaml # 生产环境配置 ├── data/ # 数据层 │ ├── dataloader/ # 数据加载器 │ │ ├── base_dataloader.py │ │ ├── csv_dataloader.py │ │ └── database_dataloader.py │ ├── database/ # 数据库操作可选 │ └── utils/ # 数据工具函数 ├── strategy/ # 策略层 │ ├── base.py # 策略基类 │ ├── indicators/ # 技术指标库 │ └── examples/ # 示例策略 │ ├── moving_average_cross.py │ └── rsi_strategy.py ├── execution/ # 执行层 │ ├── backtest/ # 回测引擎 │ │ ├── engine.py │ │ └── broker.py # 模拟经纪商 │ ├── live/ # 实盘交易按需 │ │ └── gateway.py # 交易网关接口 │ └── portfolio.py # 投资组合管理 ├── analysis/ # 分析层 │ ├── performance.py # 绩效分析 │ └── visualizer.py # 可视化 ├── tasks/ # 应用层任务定义 │ ├── backtest_task.py │ └── optimize_task.py ├── utils/ # 通用工具 │ ├── logger.py # 日志配置 │ └── scheduler.py # 任务调度可选 ├── tests/ # 单元测试 ├── logs/ # 日志目录运行时生成 ├── results/ # 回测结果目录运行时生成 ├── requirements.txt # Python依赖 └── main.py # 主程序入口关键模块设计要点策略基类这是框架的“宪法”。它定义了所有策略必须实现的方法如on_bar,on_tick并规定了策略与执行层交互的协议。一个简单的策略基类可能长这样# strategy/base.py from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Dict, Any, List import pandas as pd dataclass class Signal: symbol: str action: str # BUY, SELL, HOLD price: float quantity: float class BaseStrategy(ABC): def __init__(self, config: Dict[str, Any]): self.config config self.initialized False def init(self): 策略初始化加载数据或计算指标 pass abstractmethod def on_bar(self, bar: pd.Series) - List[Signal]: 接收一个K线数据返回交易信号列表 pass def on_tick(self, tick): 接收一个tick数据可选 pass def get_state(self): 获取策略当前状态用于持久化或恢复 return {}数据加载器提供统一接口。上层策略只需要调用dataloader.get_bars(...)而不需要关心数据是从pandas.read_csv还是sqlalchemy读出来的。回测引擎这是最复杂的模块之一。它的核心是一个事件循环按时间顺序推进在每一个时间点获取当前的市场数据。调用策略的on_bar或on_tick方法。接收策略产生的信号。将信号提交给模拟经纪商执行。更新投资组合状态。记录交易和绩效。配置管理使用yaml或.env文件管理所有可变参数。绝对不要将API密钥、数据库密码等硬编码在代码中。配置应能按环境开发/生产加载和覆盖。2.3 设计原则总结高内聚低耦合在实现上述模块时时刻牢记两个核心软件设计原则高内聚每个模块如一个数据加载器、一个策略类只做好一件事并且把这件事做好。相关的功能和数据紧密地组织在一起。低耦合模块之间通过清晰、简单的接口进行通信尽量减少相互依赖。一个模块的修改不应导致其他模块的大规模改动。Harness Engineering 就是通过框架强制推行这些原则从而提升代码质量。3. 手把手实践从零搭建一个最小可行框架理论说再多不如动手做。我们现在来搭建一个最精简但五脏俱全的量化框架它只包含最核心的链条数据 - 策略 - 回测 - 分析。这个框架的目标不是功能强大而是结构清晰可以作为你未来扩展的坚实基石。3.1 第一步初始化项目与环境首先创建项目目录并建立虚拟环境。mkdir my_quant_framework cd my_quant_framework python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装核心依赖 pip install pandas numpy matplotlib loguru pyyaml创建requirements.txt文件记录依赖。3.2 第二步实现数据层——统一的数据接口我们在data/dataloader/下创建两个文件# data/dataloader/base_dataloader.py from abc import ABC, abstractmethod import pandas as pd class BaseDataLoader(ABC): 数据加载器基类定义统一接口 abstractmethod def get_bars(self, symbol: str, start_date: str, end_date: str, freq: strD) - pd.DataFrame: 获取K线数据 :param symbol: 标的代码 :param start_date: 开始日期 YYYY-MM-DD :param end_date: 结束日期 YYYY-MM-DD :param freq: 频率如 D日, 60min60分钟 :return: DataFrame至少包含 open, high, low, close, volume 列索引为datetime pass# data/dataloader/csv_dataloader.py import pandas as pd from .base_dataloader import BaseDataLoader class CSVDataLoader(BaseDataLoader): 从CSV文件加载数据用于演示和回测 def __init__(self, data_dir: str ./data): self.data_dir data_dir def get_bars(self, symbol: str, start_date: str, end_date: str, freq: strD) - pd.DataFrame: # 假设CSV文件名为 {symbol}.csv包含日期和OHLCV列 file_path f{self.data_dir}/{symbol}.csv try: df pd.read_csv(file_path, index_coldate, parse_datesTrue) df df.loc[start_date:end_date] # 这里可以添加resample逻辑来转换频率为简化示例省略 return df[[open, high, low, close, volume]] except FileNotFoundError: raise FileNotFoundError(f数据文件 {file_path} 未找到请检查路径和文件名。)这样我们就有了一个可插拔的数据层。未来要接入数据库或API只需实现新的DataLoader子类。3.3 第三步实现策略层——一个简单的双均线策略在strategy/examples/下创建我们的第一个策略。# strategy/examples/moving_average_cross.py import pandas as pd from typing import List from ..base import BaseStrategy, Signal class MovingAverageCrossStrategy(BaseStrategy): 双均线交叉策略 def __init__(self, config): super().__init__(config) self.short_window config.get(short_window, 10) self.long_window config.get(long_window, 30) self.symbol config.get(symbol, 000001.SZ) self.position 0 # 当前持仓正数表示多头0表示空仓 def on_bar(self, bar: pd.Series) - List[Signal]: # 注意在实际框架中策略内部会维护一个数据窗口来计算指标 # 这里为简化假设bar是包含历史数据的DataFrame中的一行策略能访问到完整历史 # 实际实现中策略基类或上下文会提供历史数据访问方法 signals [] # 模拟计算如果短期均线上穿长期均线且空仓则买入 # 如果短期均线下穿长期均线且持有多头则卖出 # (此处为逻辑示意省略具体的均线计算和状态判断) action HOLD if self.position 0 and some_condition: # 假设满足买入条件 action BUY quantity 100 # 计算出的数量 elif self.position 0 and some_other_condition: # 假设满足卖出条件 action SELL quantity self.position if action ! HOLD: signals.append(Signal( symbolself.symbol, actionaction, pricebar[close], quantityquantity )) # 更新模拟持仓 self.position quantity if action BUY else 0 return signals这个策略虽然简单但它完全符合我们定义的接口。它接收市场数据bar根据内部逻辑计算返回标准的Signal对象列表。3.4 第四步实现执行层——一个简单的回测引擎核心回测引擎是框架中最复杂的部分。我们实现一个极度简化的版本展示其核心循环。# execution/backtest/engine.py import pandas as pd from typing import Dict, Any, List from ...strategy.base import BaseStrategy, Signal from .broker import BacktestBroker from ...analysis.performance import calculate_metrics class BacktestEngine: def __init__(self, initial_cash: float 100000.0): self.initial_cash initial_cash self.broker BacktestBroker(initial_cash) self.strategy None self.data None self.results {} def set_strategy(self, strategy: BaseStrategy): self.strategy strategy def load_data(self, data: pd.DataFrame): 加载回测数据DataFrame索引应为datetime self.data data def run(self): if self.strategy is None or self.data is None: raise ValueError(请先设置策略和加载数据) print(开始回测...) # 策略初始化 self.strategy.init() # 按时间顺序遍历数据 for idx, bar in self.data.iterrows(): # 更新经纪商时间用于检查订单是否成交、计算持仓市值等 self.broker.update_time(idx) # 1. 获取策略信号 signals: List[Signal] self.strategy.on_bar(bar) # 2. 提交订单到模拟经纪商 for signal in signals: self.broker.submit_order(signal) # 3. 经纪商处理订单模拟撮合 self.broker.match_orders(bar) # 4. 记录当前快照可选用于后续分析 self.broker.record_snapshot(idx, bar[close]) print(回测结束。) # 计算绩效 equity_curve self.broker.get_equity_curve() self.results[equity_curve] equity_curve self.results[final_portfolio_value] self.broker.get_portfolio_value() self.results[trades] self.broker.get_trade_log() self.results[metrics] calculate_metrics(equity_curve) return self.results其中BacktestBroker负责模拟交易撮合、管理现金和持仓、记录交易日志。calculate_metrics函数根据权益曲线计算夏普比率、最大回撤等指标。3.5 第五步组装并运行——体验完整流程最后我们创建一个主程序main.py将各个模块像拼图一样组装起来。# main.py import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from data.dataloader.csv_dataloader import CSVDataLoader from strategy.examples.moving_average_cross import MovingAverageCrossStrategy from execution.backtest.engine import BacktestEngine import matplotlib.pyplot as plt def main(): # 1. 准备数据 data_loader CSVDataLoader(data_dir./sample_data) # 假设我们有一份沪深300指数的CSV数据 df data_loader.get_bars(symbol000300.SH, start_date2020-01-01, end_date2023-12-31) # 2. 创建并配置策略 strategy_config { symbol: 000300.SH, short_window: 10, long_window: 30 } strategy MovingAverageCrossStrategy(strategy_config) # 3. 创建回测引擎并运行 engine BacktestEngine(initial_cash100000) engine.set_strategy(strategy) engine.load_data(df) results engine.run() # 4. 输出结果 print(f最终资产: {results[final_portfolio_value]:.2f}) print(f夏普比率: {results[metrics].get(sharpe_ratio, N/A):.4f}) print(f最大回撤: {results[metrics].get(max_drawdown, N/A):.2%}) # 5. 简单可视化 equity_curve results[equity_curve] plt.figure(figsize(12, 6)) plt.plot(equity_curve.index, equity_curve[total], labelPortfolio Value) plt.title(Backtest Equity Curve) plt.xlabel(Date) plt.ylabel(Value) plt.legend() plt.grid(True) plt.savefig(./results/equity_curve.png) plt.show() if __name__ __main__: main()运行这个程序你就完成了一次从数据加载、策略运算到回测执行的完整闭环。虽然这个框架极其简陋但它清晰地展示了各模块如何通过接口协作为后续扩展打下了完美的基础。4. 从“能跑”到“好用”框架的进阶与工程化考量让一个框架“能跑”只是第一步。要让它在真实的研究和生产环境中“好用”我们还需要考虑很多工程化细节。这正是 Harness Engineering 发挥威力的地方——通过框架的约束和引导将这些最佳实践固化下来。4.1 配置管理环境隔离与安全千万不要在代码里写死任何配置。使用配置文件如config/default.yaml来管理一切可变参数# config/default.yaml data: source: csv # 或 database, api csv_dir: ./data database_url: sqlite:///quant.db backtest: initial_cash: 100000 commission: 0.0003 # 手续费率 slippage: 0.001 # 滑点 strategy: ma_cross: short_window: 10 long_window: 30 symbol: 000300.SH logging: level: INFO file: ./logs/backtest.log在主程序中使用像PyYAML这样的库来加载配置并支持环境覆盖如config/production.yaml覆盖config/default.yaml中的生产环境特定值。注意涉及API密钥、数据库密码等敏感信息应使用环境变量或专门的密钥管理服务绝对不要提交到版本库中。可以在配置中引用环境变量如api_key: ${MY_API_KEY}。4.2 日志与可观测性知道系统在做什么日志是调试和监控的生命线。不要用print使用标准的日志库如 Python 的logging或更友好的loguru。在框架初始化时统一配置日志格式、级别和输出位置文件、控制台。# utils/logger.py from loguru import logger import sys def setup_logger(log_levelINFO, log_file./logs/app.log): logger.remove() # 移除默认配置 logger.add(sys.stderr, levellog_level, formatgreen{time:YYYY-MM-DD HH:mm:ss}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan:cyan{line}/cyan - level{message}/level) if log_file: logger.add(log_file, rotation10 MB, retention30 days, levellog_level) return logger # 在其他模块中引入 from utils.logger import logger logger.info(回测引擎启动) logger.warning(数据缺失进行插值处理) logger.error(订单提交失败: {}, error_msg)好的日志应该能让你在出问题时快速定位到时间、模块、函数和具体的错误信息。4.3 异常处理与健壮性应对不确定的世界网络会断数据会缺失API会限流。框架必须能优雅地处理异常而不是直接崩溃。数据层处理网络超时、数据格式错误、缺失值填充。策略层处理计算异常如除零错误确保策略状态可恢复。执行层处理订单拒绝、部分成交、连接中断并实现重试机制。任务层捕获整个回测或实盘任务中的未处理异常记录错误上下文并决定是终止任务还是跳过错误继续。# 示例在数据加载中增加重试 from tenacity import retry, stop_after_attempt, wait_exponential class APIDataLoader(BaseDataLoader): retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def get_bars(self, symbol, start_date, end_date, freq): try: # 调用API data self._call_external_api(...) return self._parse_data(data) except (RequestException, Timeout) as e: logger.error(f获取{symbol}数据失败: {e}) raise # 重试装饰器会捕获此异常并重试4.4 性能优化当数据量变大时回测成千上万只股票、使用高频数据或进行参数优化时性能会成为瓶颈。优化点包括向量化操作在策略中尽量使用pandas和numpy的向量化函数避免在数据循环中进行Python级计算。缓存对频繁读取且不变的基础数据如股票列表、复权因子进行缓存。并行计算参数优化、多标的回测等“令人尴尬的并行”任务可以使用multiprocessing或joblib进行并行化。使用更高效的数据结构对于订单簿、tick级回测考虑使用专门的数据结构。框架应该为这些优化提供可能性例如提供一个并行的参数优化任务类。4.5 版本控制与实验管理这是量化研究的核心。你的框架应该能自然地与Git等版本控制工具协作。更重要的是要能记录每次实验的“快照”代码版本Git commit hash。配置参数策略参数、回测起止时间、初始资金等。数据版本使用的数据源和日期范围。结果回测绩效指标、交易记录、权益曲线。可以将这些信息自动序列化如保存为JSON文件到每次运行的独立结果目录中。这样任何成功的策略都可以被精确复现任何失败的实验也都有据可查。5. 超越回测框架的延伸与未来一个成熟的量化框架其边界远不止于回测。Harness Engineering 的思想鼓励我们以统一的、工程化的方式看待整个量化工作流。5.1 实盘交易集成回测引擎和实盘交易网关应实现相同的接口。你的策略基类on_bar方法产生的Signal应该既能被回测引擎的模拟经纪商处理也能被实盘网关转换为真实的交易所订单。这意味着你需要抽象交易网关定义submit_order,cancel_order,get_position等标准方法。实现具体网关针对不同的券商或量化平台如CTP、盈透、RiceQuant等实现具体网关。风控模块前置在订单到达网关前必须经过严格的风控检查仓位限制、单笔最大金额、黑名单等。状态同步与监控实盘系统需要7x24小时监控订单状态、持仓、资金并具备异常报警和自动处理能力。5.2 策略研究与分析平台框架可以演变成一个内部的研究平台策略超市将策略作为可插拔的组件进行管理研究人员可以提交、测试、对比不同策略。自动化报告回测结束后自动生成包含绩效分析、图表、归因分析的标准报告。参数优化与机器学习集成超参数优化库如Optuna甚至将策略生成过程本身建模为机器学习问题如基于强化学习。5.3 从“项目”到“产品”的思维转变最终当你和你的团队依赖这套框架进行日常研究和交易时它就不再是一个“项目”而是一个内部“产品”。这意味着你需要完善的文档包括架构说明、API文档、部署指南、故障排查。持续集成/持续部署自动化测试、代码质量检查、自动化部署到生产服务器。监控与告警对实盘系统的关键指标延迟、成交率、错误率进行监控。团队协作流程定义策略从研究、回测、模拟盘到实盘的完整上线流程和评审机制。搭建量化框架的旅程本质上是一个将量化交易从“手工作坊”升级为“现代软件工程”的过程。Harness Engineering 提供的不是一条捷径而是一张经过验证的蓝图和一套严谨的施工规范。它要求你在开始时投入更多精力去设计、去约束、去抽象但这份投入会在项目复杂度增长时以指数级的方式回报你——以更少的错误、更快的迭代速度和更从容的协作体验。现在你已经拥有了这张蓝图和第一块基石。接下来要做的就是在你自己的数据和想法之上开始搭建并在实践中不断重构和完善它。记住最好的框架不是一开始就设计完美的而是在解决真实问题的过程中逐渐演化出来的。 量化交易工程化框架：从策略脚本到可复用系统的构建指南
最近在和一些做量化交易的朋友聊天发现一个挺有意思的现象很多人花了大把时间研究策略、调参、回测但真正把策略从“能跑”变成“能稳定、可维护地跑”这一步却总是磕磕绊绊。策略代码、数据源、回测引擎、风控模块、日志系统……这些组件像一堆散落的乐高积木每次想换个策略或者加个新数据源都得重新“搭”一遍过程繁琐不说还容易出错。这让我想起了软件工程里一个老生常谈的问题如何把一次性的、临时的脚本变成一套可复用、可迭代、可协作的工程化系统。在量化领域这个问题尤其突出。你可能会用backtrader、zipline或者自己写一套回测但如何管理不同策略的版本如何确保生产环境和回测环境一致如何优雅地处理实时数据流和批量历史数据如何把策略部署到实盘时还能方便地监控和干预这时候一个清晰的工程化框架就显得至关重要。它不直接帮你赚钱但它能让你赚钱的工具变得更可靠、更高效。今天我们不谈具体的阿尔法策略而是聚焦于如何“搭建”这个工具本身。我们将围绕Harness Engineering这个理念手把手构建一个属于你自己的、可扩展的量化框架。这里的“Harness”不是某个特定软件而是一种“约束与引导”的工程思想——通过一套约定、工具和流程把散乱的量化开发活动“套”进一个高效、规范的轨道里。1. 为什么你的量化策略需要一个“工程化框架”在深入动手之前我们必须先达成一个共识为什么不能直接写个脚本就跑工程化框架到底解决了什么核心痛点很多人对框架有抵触觉得是“过度设计”增加了学习成本。但事实上缺乏框架的“自由”在项目稍微复杂一点后会带来更大的混乱成本。1.1 从“一次性实验”到“可复用资产”的转变量化策略开发天然具有实验性质。你可能会尝试几十个想法最终只有一两个能进入实盘。如果没有框架每个实验可能都是一个独立的、结构各异的脚本。时间一长你会发现代码无法复用A策略的数据处理逻辑写得很好但B策略无法直接调用因为数据加载、清洗的代码和策略逻辑紧耦合在一起。结果无法复现三个月后你发现某个曾经有效的策略失效了想回头检查当时的回测结果和参数却发现找不到当时的代码版本、数据快照和环境配置。协作成为噩梦当你需要和同伴一起开发时光是统一代码风格、定义数据接口、约定回测流程就要耗费大量沟通成本。一个工程化框架的核心价值就是帮你把每一次实验的“副产品”——比如数据获取模块、回测引擎接口、绩效分析工具——沉淀下来变成团队共享的、标准化的“资产”。下一次新策略开发你只需要关心策略逻辑本身其他“轮子”都是现成的、经过测试的。1.2 管理复杂性策略、数据、执行的三重奏一个完整的量化系统至少包含三个核心维度策略产生交易信号的逻辑。它可能是基于技术指标、基本面数据、机器学习模型甚至是另类数据。数据策略的“燃料”。包括历史数据用于回测和实时数据用于实盘数据源可能来自数据库、CSV文件、API接口格式和频率各异。执行如何将策略信号转化为实际的交易。这包括回测时的模拟执行以及实盘时的真实订单执行。如果没有框架这三者通常会混杂在一起。一个脚本里可能既有关闭数据库连接又有计算移动平均线还有打印交易日志。这种混杂使得任何一部分的修改都风险极高。框架的作用就是强制性地进行关注点分离通过清晰的接口和模块边界让策略开发者、数据工程师、运维人员可以相对独立地工作。1.3 Harness Engineering不是工具而是“约束的艺术”“Harness”这个词原意是马具引申为“利用并控制”。Harness Engineering 的精髓在于通过设计一套合理的约束框架、规范、工具链来释放开发者的生产力并保障系统的可靠性。它不是一个现成的软件虽然有很多工具可以辅助而是一种设计思想。对于量化框架来说这意味着约束项目结构强制规定代码、配置、数据、日志的存放位置。所有人都按同一套模板开始新人上手极快。约束数据流定义清晰的数据接口Input/Output所有模块都通过标准接口交换数据避免隐式依赖。约束执行流程将回测、优化、实盘部署定义为一个个可配置的“任务”或“流水线”使其可重复、可监控。约束配置管理将策略参数、数据库连接信息、API密钥等与环境相关的配置集中管理并与代码分离便于不同环境开发、测试、生产的切换。这种“约束”看似限制了自由实则提供了更大的自由——在框架划定的安全边界内你可以专注于创造性的策略研究而不用为工程细节分心。2. 搭建量化框架的核心模块与设计原则理解了“为什么”我们来看“是什么”。一个典型的、基于 Harness Engineering 思想的量化框架应该包含哪些核心模块它们之间如何协作这里我们抽象出一个通用的分层架构。2.1 分层架构从数据到决策的清晰流水线一个健壮的框架通常采用分层设计每一层职责单一上层依赖下层提供的服务。一个经典的量化框架可以划分为以下四层层级核心职责关键模块举例设计原则数据层获取、存储、提供标准化数据数据获取器、数据清洗器、数据存储器、数据查询接口接口标准化为上层提供统一的数据访问接口如get_bars(symbol, start_date, end_date, frequency)隐藏底层数据源数据库、CSV、API的差异。可插拔方便更换或增加新的数据源。策略层生成交易信号策略基类、技术指标库、信号计算引擎纯逻辑策略类只负责接收数据、计算信号不关心数据从哪里来、信号发到哪里去。可回测策略逻辑必须能在历史数据上独立运行用于回测。执行层模拟或真实执行交易管理投资组合状态回测引擎、实盘交易网关、投资组合管理器、风险控制器模拟与实盘统一回测引擎和实盘交易网关应实现相同的执行接口使得策略无需修改即可在两种模式下运行。状态管理精确记录现金、持仓、交易记录、绩效等状态。应用层组织工作流提供用户界面回测任务运行器、参数优化器、可视化模块、监控告警模块任务化将一次完整的回测或实盘运行定义为一个可配置、可重复执行的任务。可观测提供丰富的日志、指标和可视化结果便于分析和调试。这个分层架构是 Harness Engineering 思想的具体体现。每一层都通过明确的接口为上层提供服务同时受到下层接口的约束。修改数据源不会影响策略逻辑更换券商API也只需改动执行层的网关模块。2.2 核心模块详解与代码结构让我们把这些抽象的概念落地到一个具体的项目目录结构中。以下是一个建议的框架项目布局quant_framework/ ├── config/ # 配置管理 │ ├── default.yaml # 默认配置 │ ├── development.yaml # 开发环境配置 │ └── production.yaml # 生产环境配置 ├── data/ # 数据层 │ ├── dataloader/ # 数据加载器 │ │ ├── base_dataloader.py │ │ ├── csv_dataloader.py │ │ └── database_dataloader.py │ ├── database/ # 数据库操作可选 │ └── utils/ # 数据工具函数 ├── strategy/ # 策略层 │ ├── base.py # 策略基类 │ ├── indicators/ # 技术指标库 │ └── examples/ # 示例策略 │ ├── moving_average_cross.py │ └── rsi_strategy.py ├── execution/ # 执行层 │ ├── backtest/ # 回测引擎 │ │ ├── engine.py │ │ └── broker.py # 模拟经纪商 │ ├── live/ # 实盘交易按需 │ │ └── gateway.py # 交易网关接口 │ └── portfolio.py # 投资组合管理 ├── analysis/ # 分析层 │ ├── performance.py # 绩效分析 │ └── visualizer.py # 可视化 ├── tasks/ # 应用层任务定义 │ ├── backtest_task.py │ └── optimize_task.py ├── utils/ # 通用工具 │ ├── logger.py # 日志配置 │ └── scheduler.py # 任务调度可选 ├── tests/ # 单元测试 ├── logs/ # 日志目录运行时生成 ├── results/ # 回测结果目录运行时生成 ├── requirements.txt # Python依赖 └── main.py # 主程序入口关键模块设计要点策略基类这是框架的“宪法”。它定义了所有策略必须实现的方法如on_bar,on_tick并规定了策略与执行层交互的协议。一个简单的策略基类可能长这样# strategy/base.py from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Dict, Any, List import pandas as pd dataclass class Signal: symbol: str action: str # BUY, SELL, HOLD price: float quantity: float class BaseStrategy(ABC): def __init__(self, config: Dict[str, Any]): self.config config self.initialized False def init(self): 策略初始化加载数据或计算指标 pass abstractmethod def on_bar(self, bar: pd.Series) - List[Signal]: 接收一个K线数据返回交易信号列表 pass def on_tick(self, tick): 接收一个tick数据可选 pass def get_state(self): 获取策略当前状态用于持久化或恢复 return {}数据加载器提供统一接口。上层策略只需要调用dataloader.get_bars(...)而不需要关心数据是从pandas.read_csv还是sqlalchemy读出来的。回测引擎这是最复杂的模块之一。它的核心是一个事件循环按时间顺序推进在每一个时间点获取当前的市场数据。调用策略的on_bar或on_tick方法。接收策略产生的信号。将信号提交给模拟经纪商执行。更新投资组合状态。记录交易和绩效。配置管理使用yaml或.env文件管理所有可变参数。绝对不要将API密钥、数据库密码等硬编码在代码中。配置应能按环境开发/生产加载和覆盖。2.3 设计原则总结高内聚低耦合在实现上述模块时时刻牢记两个核心软件设计原则高内聚每个模块如一个数据加载器、一个策略类只做好一件事并且把这件事做好。相关的功能和数据紧密地组织在一起。低耦合模块之间通过清晰、简单的接口进行通信尽量减少相互依赖。一个模块的修改不应导致其他模块的大规模改动。Harness Engineering 就是通过框架强制推行这些原则从而提升代码质量。3. 手把手实践从零搭建一个最小可行框架理论说再多不如动手做。我们现在来搭建一个最精简但五脏俱全的量化框架它只包含最核心的链条数据 - 策略 - 回测 - 分析。这个框架的目标不是功能强大而是结构清晰可以作为你未来扩展的坚实基石。3.1 第一步初始化项目与环境首先创建项目目录并建立虚拟环境。mkdir my_quant_framework cd my_quant_framework python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装核心依赖 pip install pandas numpy matplotlib loguru pyyaml创建requirements.txt文件记录依赖。3.2 第二步实现数据层——统一的数据接口我们在data/dataloader/下创建两个文件# data/dataloader/base_dataloader.py from abc import ABC, abstractmethod import pandas as pd class BaseDataLoader(ABC): 数据加载器基类定义统一接口 abstractmethod def get_bars(self, symbol: str, start_date: str, end_date: str, freq: strD) - pd.DataFrame: 获取K线数据 :param symbol: 标的代码 :param start_date: 开始日期 YYYY-MM-DD :param end_date: 结束日期 YYYY-MM-DD :param freq: 频率如 D日, 60min60分钟 :return: DataFrame至少包含 open, high, low, close, volume 列索引为datetime pass# data/dataloader/csv_dataloader.py import pandas as pd from .base_dataloader import BaseDataLoader class CSVDataLoader(BaseDataLoader): 从CSV文件加载数据用于演示和回测 def __init__(self, data_dir: str ./data): self.data_dir data_dir def get_bars(self, symbol: str, start_date: str, end_date: str, freq: strD) - pd.DataFrame: # 假设CSV文件名为 {symbol}.csv包含日期和OHLCV列 file_path f{self.data_dir}/{symbol}.csv try: df pd.read_csv(file_path, index_coldate, parse_datesTrue) df df.loc[start_date:end_date] # 这里可以添加resample逻辑来转换频率为简化示例省略 return df[[open, high, low, close, volume]] except FileNotFoundError: raise FileNotFoundError(f数据文件 {file_path} 未找到请检查路径和文件名。)这样我们就有了一个可插拔的数据层。未来要接入数据库或API只需实现新的DataLoader子类。3.3 第三步实现策略层——一个简单的双均线策略在strategy/examples/下创建我们的第一个策略。# strategy/examples/moving_average_cross.py import pandas as pd from typing import List from ..base import BaseStrategy, Signal class MovingAverageCrossStrategy(BaseStrategy): 双均线交叉策略 def __init__(self, config): super().__init__(config) self.short_window config.get(short_window, 10) self.long_window config.get(long_window, 30) self.symbol config.get(symbol, 000001.SZ) self.position 0 # 当前持仓正数表示多头0表示空仓 def on_bar(self, bar: pd.Series) - List[Signal]: # 注意在实际框架中策略内部会维护一个数据窗口来计算指标 # 这里为简化假设bar是包含历史数据的DataFrame中的一行策略能访问到完整历史 # 实际实现中策略基类或上下文会提供历史数据访问方法 signals [] # 模拟计算如果短期均线上穿长期均线且空仓则买入 # 如果短期均线下穿长期均线且持有多头则卖出 # (此处为逻辑示意省略具体的均线计算和状态判断) action HOLD if self.position 0 and some_condition: # 假设满足买入条件 action BUY quantity 100 # 计算出的数量 elif self.position 0 and some_other_condition: # 假设满足卖出条件 action SELL quantity self.position if action ! HOLD: signals.append(Signal( symbolself.symbol, actionaction, pricebar[close], quantityquantity )) # 更新模拟持仓 self.position quantity if action BUY else 0 return signals这个策略虽然简单但它完全符合我们定义的接口。它接收市场数据bar根据内部逻辑计算返回标准的Signal对象列表。3.4 第四步实现执行层——一个简单的回测引擎核心回测引擎是框架中最复杂的部分。我们实现一个极度简化的版本展示其核心循环。# execution/backtest/engine.py import pandas as pd from typing import Dict, Any, List from ...strategy.base import BaseStrategy, Signal from .broker import BacktestBroker from ...analysis.performance import calculate_metrics class BacktestEngine: def __init__(self, initial_cash: float 100000.0): self.initial_cash initial_cash self.broker BacktestBroker(initial_cash) self.strategy None self.data None self.results {} def set_strategy(self, strategy: BaseStrategy): self.strategy strategy def load_data(self, data: pd.DataFrame): 加载回测数据DataFrame索引应为datetime self.data data def run(self): if self.strategy is None or self.data is None: raise ValueError(请先设置策略和加载数据) print(开始回测...) # 策略初始化 self.strategy.init() # 按时间顺序遍历数据 for idx, bar in self.data.iterrows(): # 更新经纪商时间用于检查订单是否成交、计算持仓市值等 self.broker.update_time(idx) # 1. 获取策略信号 signals: List[Signal] self.strategy.on_bar(bar) # 2. 提交订单到模拟经纪商 for signal in signals: self.broker.submit_order(signal) # 3. 经纪商处理订单模拟撮合 self.broker.match_orders(bar) # 4. 记录当前快照可选用于后续分析 self.broker.record_snapshot(idx, bar[close]) print(回测结束。) # 计算绩效 equity_curve self.broker.get_equity_curve() self.results[equity_curve] equity_curve self.results[final_portfolio_value] self.broker.get_portfolio_value() self.results[trades] self.broker.get_trade_log() self.results[metrics] calculate_metrics(equity_curve) return self.results其中BacktestBroker负责模拟交易撮合、管理现金和持仓、记录交易日志。calculate_metrics函数根据权益曲线计算夏普比率、最大回撤等指标。3.5 第五步组装并运行——体验完整流程最后我们创建一个主程序main.py将各个模块像拼图一样组装起来。# main.py import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from data.dataloader.csv_dataloader import CSVDataLoader from strategy.examples.moving_average_cross import MovingAverageCrossStrategy from execution.backtest.engine import BacktestEngine import matplotlib.pyplot as plt def main(): # 1. 准备数据 data_loader CSVDataLoader(data_dir./sample_data) # 假设我们有一份沪深300指数的CSV数据 df data_loader.get_bars(symbol000300.SH, start_date2020-01-01, end_date2023-12-31) # 2. 创建并配置策略 strategy_config { symbol: 000300.SH, short_window: 10, long_window: 30 } strategy MovingAverageCrossStrategy(strategy_config) # 3. 创建回测引擎并运行 engine BacktestEngine(initial_cash100000) engine.set_strategy(strategy) engine.load_data(df) results engine.run() # 4. 输出结果 print(f最终资产: {results[final_portfolio_value]:.2f}) print(f夏普比率: {results[metrics].get(sharpe_ratio, N/A):.4f}) print(f最大回撤: {results[metrics].get(max_drawdown, N/A):.2%}) # 5. 简单可视化 equity_curve results[equity_curve] plt.figure(figsize(12, 6)) plt.plot(equity_curve.index, equity_curve[total], labelPortfolio Value) plt.title(Backtest Equity Curve) plt.xlabel(Date) plt.ylabel(Value) plt.legend() plt.grid(True) plt.savefig(./results/equity_curve.png) plt.show() if __name__ __main__: main()运行这个程序你就完成了一次从数据加载、策略运算到回测执行的完整闭环。虽然这个框架极其简陋但它清晰地展示了各模块如何通过接口协作为后续扩展打下了完美的基础。4. 从“能跑”到“好用”框架的进阶与工程化考量让一个框架“能跑”只是第一步。要让它在真实的研究和生产环境中“好用”我们还需要考虑很多工程化细节。这正是 Harness Engineering 发挥威力的地方——通过框架的约束和引导将这些最佳实践固化下来。4.1 配置管理环境隔离与安全千万不要在代码里写死任何配置。使用配置文件如config/default.yaml来管理一切可变参数# config/default.yaml data: source: csv # 或 database, api csv_dir: ./data database_url: sqlite:///quant.db backtest: initial_cash: 100000 commission: 0.0003 # 手续费率 slippage: 0.001 # 滑点 strategy: ma_cross: short_window: 10 long_window: 30 symbol: 000300.SH logging: level: INFO file: ./logs/backtest.log在主程序中使用像PyYAML这样的库来加载配置并支持环境覆盖如config/production.yaml覆盖config/default.yaml中的生产环境特定值。注意涉及API密钥、数据库密码等敏感信息应使用环境变量或专门的密钥管理服务绝对不要提交到版本库中。可以在配置中引用环境变量如api_key: ${MY_API_KEY}。4.2 日志与可观测性知道系统在做什么日志是调试和监控的生命线。不要用print使用标准的日志库如 Python 的logging或更友好的loguru。在框架初始化时统一配置日志格式、级别和输出位置文件、控制台。# utils/logger.py from loguru import logger import sys def setup_logger(log_levelINFO, log_file./logs/app.log): logger.remove() # 移除默认配置 logger.add(sys.stderr, levellog_level, formatgreen{time:YYYY-MM-DD HH:mm:ss}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan:cyan{line}/cyan - level{message}/level) if log_file: logger.add(log_file, rotation10 MB, retention30 days, levellog_level) return logger # 在其他模块中引入 from utils.logger import logger logger.info(回测引擎启动) logger.warning(数据缺失进行插值处理) logger.error(订单提交失败: {}, error_msg)好的日志应该能让你在出问题时快速定位到时间、模块、函数和具体的错误信息。4.3 异常处理与健壮性应对不确定的世界网络会断数据会缺失API会限流。框架必须能优雅地处理异常而不是直接崩溃。数据层处理网络超时、数据格式错误、缺失值填充。策略层处理计算异常如除零错误确保策略状态可恢复。执行层处理订单拒绝、部分成交、连接中断并实现重试机制。任务层捕获整个回测或实盘任务中的未处理异常记录错误上下文并决定是终止任务还是跳过错误继续。# 示例在数据加载中增加重试 from tenacity import retry, stop_after_attempt, wait_exponential class APIDataLoader(BaseDataLoader): retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def get_bars(self, symbol, start_date, end_date, freq): try: # 调用API data self._call_external_api(...) return self._parse_data(data) except (RequestException, Timeout) as e: logger.error(f获取{symbol}数据失败: {e}) raise # 重试装饰器会捕获此异常并重试4.4 性能优化当数据量变大时回测成千上万只股票、使用高频数据或进行参数优化时性能会成为瓶颈。优化点包括向量化操作在策略中尽量使用pandas和numpy的向量化函数避免在数据循环中进行Python级计算。缓存对频繁读取且不变的基础数据如股票列表、复权因子进行缓存。并行计算参数优化、多标的回测等“令人尴尬的并行”任务可以使用multiprocessing或joblib进行并行化。使用更高效的数据结构对于订单簿、tick级回测考虑使用专门的数据结构。框架应该为这些优化提供可能性例如提供一个并行的参数优化任务类。4.5 版本控制与实验管理这是量化研究的核心。你的框架应该能自然地与Git等版本控制工具协作。更重要的是要能记录每次实验的“快照”代码版本Git commit hash。配置参数策略参数、回测起止时间、初始资金等。数据版本使用的数据源和日期范围。结果回测绩效指标、交易记录、权益曲线。可以将这些信息自动序列化如保存为JSON文件到每次运行的独立结果目录中。这样任何成功的策略都可以被精确复现任何失败的实验也都有据可查。5. 超越回测框架的延伸与未来一个成熟的量化框架其边界远不止于回测。Harness Engineering 的思想鼓励我们以统一的、工程化的方式看待整个量化工作流。5.1 实盘交易集成回测引擎和实盘交易网关应实现相同的接口。你的策略基类on_bar方法产生的Signal应该既能被回测引擎的模拟经纪商处理也能被实盘网关转换为真实的交易所订单。这意味着你需要抽象交易网关定义submit_order,cancel_order,get_position等标准方法。实现具体网关针对不同的券商或量化平台如CTP、盈透、RiceQuant等实现具体网关。风控模块前置在订单到达网关前必须经过严格的风控检查仓位限制、单笔最大金额、黑名单等。状态同步与监控实盘系统需要7x24小时监控订单状态、持仓、资金并具备异常报警和自动处理能力。5.2 策略研究与分析平台框架可以演变成一个内部的研究平台策略超市将策略作为可插拔的组件进行管理研究人员可以提交、测试、对比不同策略。自动化报告回测结束后自动生成包含绩效分析、图表、归因分析的标准报告。参数优化与机器学习集成超参数优化库如Optuna甚至将策略生成过程本身建模为机器学习问题如基于强化学习。5.3 从“项目”到“产品”的思维转变最终当你和你的团队依赖这套框架进行日常研究和交易时它就不再是一个“项目”而是一个内部“产品”。这意味着你需要完善的文档包括架构说明、API文档、部署指南、故障排查。持续集成/持续部署自动化测试、代码质量检查、自动化部署到生产服务器。监控与告警对实盘系统的关键指标延迟、成交率、错误率进行监控。团队协作流程定义策略从研究、回测、模拟盘到实盘的完整上线流程和评审机制。搭建量化框架的旅程本质上是一个将量化交易从“手工作坊”升级为“现代软件工程”的过程。Harness Engineering 提供的不是一条捷径而是一张经过验证的蓝图和一套严谨的施工规范。它要求你在开始时投入更多精力去设计、去约束、去抽象但这份投入会在项目复杂度增长时以指数级的方式回报你——以更少的错误、更快的迭代速度和更从容的协作体验。现在你已经拥有了这张蓝图和第一块基石。接下来要做的就是在你自己的数据和想法之上开始搭建并在实践中不断重构和完善它。记住最好的框架不是一开始就设计完美的而是在解决真实问题的过程中逐渐演化出来的。