# Python 量化回测实战：从零搭建双均线策略回测系统（免费数据源 + 完整代码）

> 量化交易的核心不是找到一个"神奇策略"，而是用历史数据验证策略是否可行。本文不依赖任何回测框架，用纯 Python + pandas 从零搭建一个完整的双均线策略回测系统，包含信号生成、收益计算、最大回撤、年化收益等核心指标。

---

## 一、回测到底在做什么？

回测就是拿历史数据模拟交易，回答一个问题：

> **如果过去按这个策略操作，能赚多少钱？**

一个最小可用的回测系统需要：

1. 历史 K 线数据
2. 交易信号生成逻辑
3. 模拟买卖过程
4. 收益与风险指标计算

本文选择最经典的**双均线策略**作为示例：

- **MA5 上穿 MA20** -> 买入
- **MA5 下穿 MA20** -> 卖出

---

## 二、环境准备

### 安装依赖

```bash
pip install "tickflow[all]" --upgrade
```

只需要一个库就够了，TickFlow SDK 自带 pandas 支持。

---

## 三、获取历史数据

使用 TickFlow 的免费服务获取日 K 线，无需注册：

```python
from tickflow import TickFlow

tf = TickFlow.free()

df = tf.klines.get(
    "600519.SH",
    period="1d",
    count=5000,
    adjust="forward",
    as_dataframe=True,
)

print(f"获取到 {len(df)} 根 K 线")
print(df[["trade_date", "open", "high", "low", "close", "volume"]].tail())
```

这里选择贵州茅台（600519.SH）作为示例，获取尽可能多的历史数据，并使用前复权以保证价格连续性。

---

## 四、生成交易信号

### 计算均线

```python
df["ma5"] = df["close"].rolling(5).mean()
df["ma20"] = df["close"].rolling(20).mean()
```

### 生成买卖信号

```python
df["signal"] = 0

# MA5 上穿 MA20 -> 买入信号
buy_mask = (df["ma5"] > df["ma20"]) & (df["ma5"].shift(1) <= df["ma20"].shift(1))
df.loc[buy_mask, "signal"] = 1

# MA5 下穿 MA20 -> 卖出信号
sell_mask = (df["ma5"] < df["ma20"]) & (df["ma5"].shift(1) >= df["ma20"].shift(1))
df.loc[sell_mask, "signal"] = -1

# 查看信号
signals = df[df["signal"] != 0][["trade_date", "close", "ma5", "ma20", "signal"]]
print(f"共产生 {len(signals)} 个交易信号")
print(signals.tail(10))
```

---

## 五、模拟交易过程

```python
import pandas as pd

def backtest(df, initial_capital=100000, commission=0.001):
    """
    简单回测引擎
    - initial_capital: 初始资金
    - commission: 手续费率（单边，默认千一）
    """
    capital = initial_capital
    shares = 0
    position = 0  # 0: 空仓, 1: 持仓
    trades = []

    for i, row in df.iterrows():
        if row["signal"] == 1 and position == 0:
            # 买入：用全部资金买入
            price = row["close"]
            cost = capital * commission
            shares = int((capital - cost) / price / 100) * 100  # 按手取整
            if shares <= 0:
                continue
            buy_amount = shares * price
            capital -= buy_amount + buy_amount * commission
            position = 1
            trades.append({
                "date": row["trade_date"],
                "action": "BUY",
                "price": price,
                "shares": shares,
                "capital": capital,
            })

        elif row["signal"] == -1 and position == 1:
            # 卖出：卖出全部持仓
            price = row["close"]
            sell_amount = shares * price
            capital += sell_amount - sell_amount * commission
            trades.append({
                "date": row["trade_date"],
                "action": "SELL",
                "price": price,
                "shares": shares,
                "capital": capital,
            })
            shares = 0
            position = 0

    # 如果最后还持仓，按最后一天收盘价结算
    if position == 1:
        last_price = df.iloc[-1]["close"]
        sell_amount = shares * last_price
        capital += sell_amount - sell_amount * commission
        trades.append({
            "date": df.iloc[-1]["trade_date"],
            "action": "SELL(END)",
            "price": last_price,
            "shares": shares,
            "capital": capital,
        })

    return capital, trades


final_capital, trades = backtest(df)
trade_df = pd.DataFrame(trades)

print(f"\n初始资金: 100,000")
print(f"最终资金: {final_capital:,.2f}")
print(f"总收益率: {(final_capital / 100000 - 1) * 100:.2f}%")
print(f"交易次数: {len(trade_df)}")
print("\n最近 10 笔交易:")
print(trade_df.tail(10).to_string(index=False))
```

---

## 六、计算核心风险指标

一个回测系统不能只看收益率，还需要看风险指标。

### 日收益率序列

```python
# 计算策略的每日收益率
df["position"] = 0
pos = 0
for i in range(len(df)):
    if df.iloc[i]["signal"] == 1:
        pos = 1
    elif df.iloc[i]["signal"] == -1:
        pos = 0
    df.iloc[i, df.columns.get_loc("position")] = pos

df["strategy_return"] = df["close"].pct_change() * df["position"].shift(1)
df["cumulative_return"] = (1 + df["strategy_return"]).cumprod()
df["benchmark_return"] = df["close"] / df["close"].iloc[0]
```

### 最大回撤

```python
def max_drawdown(cumulative_returns):
    peak = cumulative_returns.expanding().max()
    drawdown = (cumulative_returns - peak) / peak
    return drawdown.min()

mdd = max_drawdown(df["cumulative_return"].dropna())
print(f"最大回撤: {mdd * 100:.2f}%")
```

### 年化收益率

```python
total_days = len(df)
trading_days_per_year = 244
total_return = df["cumulative_return"].iloc[-1] - 1
years = total_days / trading_days_per_year
annual_return = (1 + total_return) ** (1 / years) - 1

print(f"年化收益率: {annual_return * 100:.2f}%")
```

### 夏普比率

```python
import numpy as np

risk_free_rate = 0.02
daily_rf = risk_free_rate / trading_days_per_year
excess_returns = df["strategy_return"].dropna() - daily_rf
sharpe = np.sqrt(trading_days_per_year) * excess_returns.mean() / excess_returns.std()

print(f"夏普比率: {sharpe:.2f}")
```

### 汇总输出

```python
print("=" * 50)
print("回测结果汇总")
print("=" * 50)
print(f"标的: 600519.SH (贵州茅台)")
print(f"策略: 双均线 MA5/MA20")
print(f"回测区间: {df['trade_date'].iloc[0]} ~ {df['trade_date'].iloc[-1]}")
print(f"交易天数: {total_days}")
print(f"总收益率: {total_return * 100:.2f}%")
print(f"年化收益率: {annual_return * 100:.2f}%")
print(f"最大回撤: {mdd * 100:.2f}%")
print(f"夏普比率: {sharpe:.2f}")
print(f"交易次数: {len(trades)}")
print("=" * 50)
```

---

## 七、多股票批量回测

真正的量化研究不会只跑一只股票。下面对多只股票批量回测：

```python
from tickflow import TickFlow
import pandas as pd
import numpy as np

tf = TickFlow.free()

test_symbols = [
    "600519.SH",  # 贵州茅台
    "000858.SZ",  # 五粮液
    "601318.SH",  # 中国平安
    "600036.SH",  # 招商银行
    "000001.SZ",  # 平安银行
]

dfs = tf.klines.batch(
    test_symbols,
    period="1d",
    count=5000,
    adjust="forward",
    as_dataframe=True,
    show_progress=True,
)

results = []

for symbol, df in dfs.items():
    if len(df) < 60:
        continue

    # 计算均线
    df["ma5"] = df["close"].rolling(5).mean()
    df["ma20"] = df["close"].rolling(20).mean()

    # 生成信号
    df["signal"] = 0
    buy = (df["ma5"] > df["ma20"]) & (df["ma5"].shift(1) <= df["ma20"].shift(1))
    sell = (df["ma5"] < df["ma20"]) & (df["ma5"].shift(1) >= df["ma20"].shift(1))
    df.loc[buy, "signal"] = 1
    df.loc[sell, "signal"] = -1

    # 计算收益
    pos = 0
    positions = []
    for _, row in df.iterrows():
        if row["signal"] == 1:
            pos = 1
        elif row["signal"] == -1:
            pos = 0
        positions.append(pos)
    df["position"] = positions
    df["strategy_return"] = df["close"].pct_change() * pd.Series(positions).shift(1).values

    cum_ret = (1 + df["strategy_return"].fillna(0)).cumprod()
    total_return = cum_ret.iloc[-1] - 1

    # 最大回撤
    peak = cum_ret.expanding().max()
    mdd = ((cum_ret - peak) / peak).min()

    # 年化
    years = len(df) / 244
    annual = (1 + total_return) ** (1 / years) - 1 if years > 0 else 0

    results.append({
        "symbol": symbol,
        "total_return": f"{total_return * 100:.2f}%",
        "annual_return": f"{annual * 100:.2f}%",
        "max_drawdown": f"{mdd * 100:.2f}%",
        "trade_days": len(df),
    })

result_df = pd.DataFrame(results)
print(result_df.to_string(index=False))
```

---

## 八、策略改进方向

基础的双均线策略只是起点，可以在此基础上做很多改进：

### 1. 增加过滤条件

```python
# 在买入信号基础上，增加成交量放大条件
df["vol_ma20"] = df["volume"].rolling(20).mean()
buy_with_volume = buy & (df["volume"] > df["vol_ma20"] * 1.5)
```

### 2. 加入止损

```python
# 买入后跌破买入价的 5% 则止损
stop_loss_pct = 0.05
```

### 3. 使用不同周期

```python
# 周K回测
df_weekly = tf.klines.get("600519.SH", period="1w", count=1000, adjust="forward", as_dataframe=True)
```

### 4. 换用其他指标

可以把均线换成 MACD 金叉/死叉、RSI 超买超卖、布林带突破等。数据获取方式完全一样，只需修改信号生成逻辑。

---

## 九、总结

本文从零搭建了一个完整的量化回测系统：

1. **数据获取**：TickFlow 免费层获取全量历史日 K 线
2. **信号生成**：双均线交叉策略
3. **模拟交易**：含手续费计算和仓位管理
4. **风险指标**：总收益率、年化收益率、最大回撤、夏普比率
5. **批量回测**：对多只股票进行对比分析

整个过程不依赖任何回测框架（如 backtrader、zipline），只用 pandas，代码清晰且容易修改。

数据是回测的基础。选择一个稳定且免费的数据源，可以让你把精力集中在策略本身。

---

## 相关链接

- 官网：[https://tickflow.org](https://tickflow.org?utm_source=tencent_dev)
- 文档：[https://docs.tickflow.org](https://docs.tickflow.org)
- Github：[https://github.com/tickflow-org/tickflow](https://github.com/tickflow-org/tickflow)

---

> 回测不是为了证明策略有多好，而是为了知道它在什么条件下会失效。


量化交易的核心不是找到一个"神奇策略"，而是用历史数据验证策略是否可行。本文不依赖任何回测框架，用纯 Python + pandas 从零搭建一个完整的双均线策略回测系统，包含信号生成、收益计算、最大回撤、年化收益等核心指标。

Python 量化回测实战：从零搭建双均线策略回测系统（免费数据源 + 完整代码）

编程语言

算法

大数据

Python量化回测实战教程：从零搭建双均线策略回测系统，包含完整代码和免费数据源。详解如何使用Python+pandas实现信号生成、收益计算、最大回撤、年化收益等核心指标，适合量化交易初学者。通过贵州茅台案例演示双均线策略（MA5/MA20）的回测全过程，并扩展至多股票批量回测，提供策略改进方向。

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Python 量化回测实战：从零搭建双均线策略回测系统（免费数据源 + 完整代码）-腾讯云开发者社区-腾讯云

Python 量化回测实战：从零搭建双均线策略回测系统（免费数据源 + 完整代码）

Python 量化回测实战：从零搭建双均线策略回测系统（免费数据源 + 完整代码）

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐