Python徒手搭建股票回测框架：核心逻辑与代码解析

巷口听风

发布时间：2026-10-13| 4人浏览过

原创

本文详细介绍如何使用纯Python从零搭建一个轻量级股票回测框架。文章解析了回测类与交易类的核心职责，展示了基于Tick驱动的调度器设计，以及Context对象在数据共享中的应用。通过Hook机制实现资产统计与最大回撤计算，提供了完整的代码架构思路与GitHub项目地址，适合量化初学者参考。

回测框架概述

本文旨在通过纯Python代码实现一个股票回测框架的搭建过程。

首先明确回测框架的定义与作用。

在股票交易或量化策略开发中，验证策略可行性是必经环节。最基础且有效的方法是利用历史数据进行回测。回测框架即为此提供平台，让策略在历史数据中反复模拟交易，最终生成收益、年化收益率、最大回撤等关键指标，以此评估策略的有效性。

框架核心架构

接下来探讨回测框架的核心组成部分。

一个基本的回测框架至少应包含两个核心模块：回测类（Backtest）和交易类（Broker）。

回测类主要负责提供一系列钩子函数（Hooks），供用户植入自定义的交易逻辑；交易类则模拟市场交易平台，提供买入和卖出等具体操作方法。

代码实现细节

以下以具体代码架构为例进行说明。

该回测框架主要包含两个核心文件，目录结构如下。

backtest/ backtest.py broker.py

在backtest.py中，主要定义BackTest类以提供回测框架的核心功能，并暴露以下关键钩子函数。


def initialize(self):
  """在回测开始前的初始化"""
  pass
 def before_on_tick(self, tick):
  pass
 def after_on_tick(self, tick):
  pass
 def before_trade(self, order):
  """在交易之前会调用此函数
  可以在此放置资金管理及风险管理的代码
  如果返回True就允许交易，否则放弃交易
  """
  return True
 def on_order_ok(self, order):
  """当订单执行成功后调用"""
  pass
 def on_order_timeout(self, order):
  """当订单超时后调用"""
  pass
 def finish(self):
  """在回测结束后调用"""
  pass
 @abstractmethod
 def on_tick(self, bar):
  """
  回测实例必须实现的方法，并编写自己的交易逻辑
  """
  pass

接触过量化平台或开源回测框架的开发者对这些钩子函数应当不陌生，尽管命名可能不同，但核心功能大致一致，区别主要在于on_tick的处理方式。

这里采用on_tick而非on_bar，是因为设计意图是让交易逻辑基于每一个时间点（Tick）执行。在该时间点，策略可以获取当前及之前所有股票的数据，从而判断是否进行交易，而非仅针对单个股票逐个处理。

而在broker.py中，主要提供buy和sell两个方法用于执行交易操作。


def buy(self, code, price, shares, ttl=-1):
  """
  限价提交买入订单
  ---------
  Parameters:
   code:str
    股票代码
   price:float or None
    最高可买入的价格, 如果为None则按市价买入
   shares:int
    买入股票数量
   ttl:int
    订单允许存在的最大时间，默认为-1，永不超时
  ---------
  return:
   dict
    {
    "type": 订单类型, "buy",
    "code": 股票代码,
    "date": 提交日期,
    "ttl": 存活时间, 当ttl等于0时则超时，往后不会在执行
    "shares": 目标股份数量,
    "price": 目标价格,
    "deal_lst": 交易成功的历史数据，如
     [{"price": 成交价格,
      "date": 成交时间,
      "commission": 交易手续费,
      "shares": 成交份额
     }]
    ""
   }
  """
  if price is None:
   stock_info = self.ctx.tick_data[code]
   price = stock_info[self.deal_price]
  order = {
   "type": "buy",
   "code": code,
   "date": self.ctx.now,
   "ttl": ttl,
   "shares": shares,
   "price": price,
   "deal_lst": []
  }
  self.submit(order)
  return order
 def sell(self, code, price, shares, ttl=-1):
  """
  限价提交卖出订单
  ---------
  Parameters:
   code:str
    股票代码
   price:float or None
    最低可卖出的价格, 如果为None则按市价卖出
   shares:int
    卖出股票数量
   ttl:int
    订单允许存在的最大时间，默认为-1，永不超时
  ---------
  return:
   dict
    {
    "type": 订单类型, "sell",
    "code": 股票代码,
    "date": 提交日期,
    "ttl": 存活时间, 当ttl等于0时则超时，往后不会在执行
    "shares": 目标股份数量,
    "price": 目标价格,
    "deal_lst": 交易成功的历史数据，如
     [{"open_price": 开仓价格,
      "close_price": 成交价格,
      "close_date": 成交时间,
      "open_date": 持仓时间,
      "commission": 交易手续费,
      "shares": 成交份额,
      "profit": 交易收益}]
    ""
   }
  """
  if code not in self.position:
   return
  if price is None:
   stock_info = self.ctx.tick_data[code]
   price = stock_info[self.deal_price]
  order = {
   "type": "sell",
   "code": code,
   "date": self.ctx.now,
   "ttl": ttl,
   "shares": shares,
   "price": price,
   "deal_lst": []
  }
  self.submit(order)
  return order

为了避免过度抽象导致代码难以维护，作者倾向于使用常见的数据结构（如list、dict）而非创建过多类和方法。

在此设计中，使用一个字典（dict）来表示一个订单。

上述方法构成了回测框架的基本交易逻辑，而框架的运行还需要一个调度器（Scheduler）来持续驱动这些方法。调度器实现如下。


class Scheduler(object):
 """

调度器与上下文管理

这是整个回测过程的调度中心，通过一个个时间刻度（tick）来驱动回测逻辑的执行。

所有被调度的对象都会绑定一个名为ctx的Context对象，用于共享回测过程中的关键数据。可用变量包括。

ctx.feed: {code1: pd.DataFrame, code2: pd.DataFrame}对象 ctx.now: 循环所处时间 ctx.tick_data: 循环所处时间的所有有报价的股票报价 ctx.trade_cal: 交易日历 ctx.broker: Broker对象 ctx.bt/ctx.backtest: Backtest对象

此外，Context对象还提供以下可用方法。


 ctx.get_hist
 """
 def __init__(self):
  """"""
  self.ctx = Context()
  self._pre_hook_lst = []
  self._post_hook_lst = []
  self._runner_lst = []
 def run(self):
  # runner指存在可调用的initialize, finish, run(tick)的对象
  runner_lst = list(chain(self._pre_hook_lst, self._runner_lst, self._post_hook_lst))
  # 循环开始前为broker, backtest, hook等实例绑定ctx对象及调用其initialize方法
  for runner in runner_lst:
   runner.ctx = self.ctx
   runner.initialize()
  # 创建交易日历
  if "trade_cal" not in self.ctx:
   df = list(self.ctx.feed.values())[0]
   self.ctx["trade_cal"] = df.index
  # 通过遍历交易日历的时间依次调用runner
  # 首先调用所有pre-hook的run方法
  # 然后调用broker,backtest的run方法
  # 最后调用post-hook的run方法
  for tick in self.ctx.trade_cal:
   self.ctx.set_currnet_time(tick)
   for runner in runner_lst:
    runner.run(tick)
  # 循环结束后调用所有runner对象的finish方法
  for runner in runner_lst:
   runner.finish()

在Backtest类实例化时，会自动创建一个调度器对象。通过调用Backtest实例的start方法即可启动调度器，调度器会根据历史数据的时间戳序列，不断驱动Backtest和Broker实例的相关方法。

为了实现不同实例间的数据访问隔离，通过将一个Context对象绑定到Backtest和Broker实例上，并通过self.ctx访问共享数据。共享数据主要包括feed对象（即历史数据），其结构为一个字典，格式如下。

{code1: pd.DataFrame, code2: pd.DataFrame}

该Context对象同时也绑定了Broker和Backtest的实例，这使得数据访问接口统一。虽然这可能导致数据访问较为随意，取决于策略编写者的使用习惯，但其优势在于减少了一堆代理方法，避免了通过添加方法去访问其他对象方法的繁琐过程。

绑定及Context对象的实现代码如下。


class Context(UserDict):
 def __getattr__(self, key):
  # 让调用这可以通过索引或者属性引用皆可
  return self[key]
 def set_currnet_time(self, tick):
  self["now"] = tick
  tick_data = {}
  # 获取当前所有有报价的股票报价
  for code, hist in self["feed"].items():
   df = hist[hist.index == tick]
   if len(df) == 1:
    tick_data[code] = df.iloc[-1]
  self["tick_data"] = tick_data
 def get_hist(self, code=None):
  """如果不指定code, 获取截至到当前时间的所有股票的历史数据"""
  if code is None:
   hist = {}
   for code, hist in self["feed"].items():
    hist[code] = hist[hist.index <= self.now]
  elif code in self.feed:
   return {code: self.feed[code]}
  return hist
class Scheduler(object):
 """

这是整个回测过程中的调度中心，通过一个个时间刻度（tick）来驱动回测逻辑。

所有被调度的对象都会绑定一个叫做ctx的Context对象，由于共享整个回测过程中的关键数据。

可用变量包括。


  ctx.feed: {code1: pd.DataFrame, code2: pd.DataFrame}对象
  ctx.now: 循环所处时间
  ctx.tick_data: 循环所处时间的所有有报价的股票报价
  ctx.trade_cal: 交易日历
  ctx.broker: Broker对象
  ctx.bt/ctx.backtest: Backtest对象

可用方法。


  ctx.get_hist
 """
 def __init__(self):
  """"""
  self.ctx = Context()
  self._pre_hook_lst = []
  self._post_hook_lst = []
  self._runner_lst = []
 def add_feed(self, feed):
  self.ctx["feed"] = feed
 def add_hook(self, hook, typ="post"):
  if typ == "post" and hook not in self._post_hook_lst:
   self._post_hook_lst.append(hook)
  elif typ == "pre" and hook not in self._pre_hook_lst:
   self._pre_hook_lst.append(hook)
 def add_broker(self, broker):
  self.ctx["broker"] = broker
 def add_backtest(self, backtest):
  self.ctx["backtest"] = backtest
  # 简写
  self.ctx["bt"] = backtest
 def add_runner(self, runner):
  if runner in self._runner_lst:
   return
  self._runner_lst.append(runner)

扩展性与统计功能

为了增强框架的可扩展性，回测框架中抽象了一个Hook类。该类允许在每次回测框架调用前或调用后执行自定义逻辑，例如统计资产变化。

这里创建了一个Stat的Hook对象，专门用于统计资产变化。


class Stat(Base):
  def __init__(self):
    self._date_hist = []
    self._cash_hist = []
    self._stk_val_hist = []
    self._ast_val_hist = []
    self._returns_hist = []

  def run(self, tick):
    self._date_hist.append(tick)
    self._cash_hist.append(self.ctx.broker.cash)
    self._stk_val_hist.append(self.ctx.broker.stock_value)
    self._ast_val_hist.append(self.ctx.broker.assets_value)

  @property
  def data(self):
    df = pd.DataFrame({"cash": self._cash_hist,
              "stock_value": self._stk_val_hist,
              "assets_value": self._ast_val_hist}, index=self._date_hist)
    df.index.name = "date"
    return df

通过这些统计的数据，可以进一步计算最大回撤、年化收益率等指标。


def get_dropdown(self):
    high_val = -1
    low_val = None
    high_index = 0
    low_index = 0
    dropdown_lst = []
    dropdown_index_lst = []
    for idx, val in enumerate(self._ast_val_hist):
      if val >= high_val:
        if high_val == low_val or high_index >= low_index:
          high_val = low_val = val
          high_index = low_index = idx
          continue
        dropdown = (high_val - low_val) / high_val
        dropdown_lst.append(dropdown)
        dropdown_index_lst.append((high_index, low_index))
        high_val = low_val = val
        high_index = low_index = idx
      if low_val is None:
        low_val = val
        low_index = idx
      if val < low_val:
        low_val = val
        low_index = idx
    if low_index > high_index:
      dropdown = (high_val - low_val) / high_val
      dropdown_lst.append(dropdown)
      dropdown_index_lst.append((high_index, low_index))
    return dropdown_lst, dropdown_index_lst
  @property
  def max_dropdown(self):
    """最大回车率"""
    dropdown_lst, dropdown_index_lst = self.get_dropdown()
    if len(dropdown_lst) > 0:
      return max(dropdown_lst)
    else:
      return 0
  @property
  def annual_return(self):
    """
    年化收益率
    y = (v/c)^(D/T) - 1
    v: 最终价值
    c: 初始价值
    D: 有效投资时间(365)

注：虽然股票市场的实际交易日约为250天，但由于持有股票期间的非交易日资金无法投资其他标的，因此此处计算年化收益率时取365天。

参考来源: https://wiki.mbalib.com/zh-tw/%E5%B9%B4%E5%8C%96%E6%94%B6%E7%9B%8A%E7%8E%87


    """
    D = 365
    c = self._ast_val_hist[0]
    v = self._ast_val_hist[-1]
    days = (self._date_hist[-1] - self._date_hist[0]).days
    ret = (v / c) ** (D / days) - 1
    return ret

至此，一个满足基本需求的回测框架即告形成。

数据与报告处理

关于交易历史数据的部分。

在该回测框架中，并未集成各种获取数据的方法，因为数据获取并非回测框架必须集成的部分，只需规定数据结构即可。数据的获取方式可参考相关数据篇文档。

关于回测报告的部分。

回测报告也被放置在回测框架之外，这里编写了一个Plottter对象用于绘制部分回测指标。结果展示如下。

接下来是一个回测示例。

下面展示一个具体的回测示例代码。


import json
from backtest import BackTest
from reporter import Plotter
class MyBackTest(BackTest):
  def initialize(self):
    self.info("initialize")
  def finish(self):
    self.info("finish")
  def on_tick(self, tick):
    tick_data = self.ctx["tick_data"]
    for code, hist in tick_data.items():
      if hist["ma10"] > 1.05 * hist["ma20"]:
        self.ctx.broker.buy(code, hist.close, 500, ttl=5)
      if hist["ma10"] < hist["ma20"] and code in self.ctx.broker.position:
        self.ctx.broker.sell(code, hist.close, 200, ttl=1)
if __name__ == '__main__':
  from utils import load_hist
  feed = {}
  for code, hist in load_hist("000002.SZ"):
    # hist = hist.iloc[:100]
    hist["ma10"] = hist.close.rolling(10).mean()
    hist["ma20"] = hist.close.rolling(20).mean()
    feed[code] = hist
  mytest = MyBackTest(feed)
  mytest.start()
  order_lst = mytest.ctx.broker.order_hist_lst
  with open("report/order_hist.json", "w") as wf:
    json.dump(order_lst, wf, indent=4, default=str)
  stats = mytest.stat
  stats.data.to_csv("report/stat.csv")
  print("策略收益： {:.3f}%".format(stats.total_returns * 100))
  print("最大回彻率: {:.3f}% ".format(stats.max_dropdown * 100))
  print("年化收益: {:.3f}% ".format(stats.annual_return * 100))
  print("夏普比率: {:.3f} ".format(stats.sharpe))
  plotter = Plotter(feed, stats, order_lst)
  plotter.report("report/report.png")

项目完整地址。

https://github.com/youerning/stock_playground

总结。

以上介绍了使用Python徒手搭建股票回测框架的全过程，希望对大家有所帮助。如有任何疑问，欢迎留言，作者会及时回复。感谢大家对本站的支持！如果觉得本文对您有帮助，欢迎转载，烦请注明出处，谢谢！