给自己搭一个量化研究工作台：Codex 管代码

很多人学量化时，会把注意力放在“策略”上。

这当然重要。但如果你真的想长期做研究，只靠零散脚本是不够的。今天一个 test.py，明天一个 ma_final_v3.py，后天一个 真的最终版.ipynb，时间久了你会发现，自己根本不知道哪个结果可信。

一个更好的做法是：从一开始就给自己搭一个轻量的量化研究工作台。

它不需要很复杂，也不需要上来就做成专业系统。你只需要把数据获取、策略代码、回测结果、图表、配置文件放到固定位置，让每一次研究都能复现。

这篇文章讲一个适合个人使用的工作台结构：AlphaFeed 负责稳定获取行情数据，Codex 帮你写策略、调试、重构和补测试。

1. 为什么需要工作台

零散脚本的最大问题不是丑，而是不可复现。

你可能会遇到这些情况：

1. 策略结果不错，但忘了当时用的参数。
2. 图表保存了，但不知道对应哪段代码。
3. 换了一组股票池，旧结果和新结果混在一起。
4. 某天改了复权方式，历史回测全变了。
5. Codex 帮你改了代码，但你没记录改动原因。

量化研究要长期迭代，必须让每次实验留下痕迹。

一个轻量工作台可以这样组织：

quant-workbench/
  data/
    raw/
    processed/
  strategies/
    ma_cross.py
    etf_rotation.py
  reports/
    charts/
    runs/
  configs/
    ma_cross.json
    etf_rotation.json
  notebooks/
  main.py
  requirements.txt

这不是唯一标准，但它能帮你把事情分清楚。

2. 数据层：统一从 AlphaFeed 取数

先写一个数据模块，比如 data_loader.py：

from pathlib import Path
import pandas as pd
from alphafeed import AlphaFeed

class AlphaFeedDataLoader:
    def __init__(self, cache_dir: str = "data/raw"):
        self.af = AlphaFeed()
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)

    def load_daily_bars(
        self,
        symbol: str,
        count: int = 800,
        adjust: str = "forward",
        use_cache: bool = True,
    ) -> pd.DataFrame:
        cache_file = self.cache_dir / f"{symbol}_{count}_{adjust}.csv"

        if use_cache and cache_file.exists():
            return pd.read_csv(cache_file)

        df = self.af.klines.get(
            symbol,
            period="1d",
            count=count,
            adjust=adjust,
            to_dataframe=True,
        )
        df = df.sort_values("trade_date").reset_index(drop=True)
        df.to_csv(cache_file, index=False)
        return df

    def load_quotes(self, symbols: list[str]) -> pd.DataFrame:
        return self.af.quotes.get(symbols=symbols, to_dataframe=True)

这样做有几个好处：

1. 策略代码不直接关心 API 细节。
2. 缓存文件可以减少重复请求。
3. 复权方式、数据长度都写在函数参数里。
4. 以后如果要换数据源，只需要改数据层。

3. 策略层：每个策略写成可复用函数

比如双均线策略：

import pandas as pd

def ma_cross_signal(
    df: pd.DataFrame,
    short_window: int = 20,
    long_window: int = 60,
) -> pd.DataFrame:
    df = df.copy()
    df["ma_short"] = df["close"].rolling(short_window).mean()
    df["ma_long"] = df["close"].rolling(long_window).mean()
    df["signal"] = (df["ma_short"] > df["ma_long"]).astype(int)
    return df

注意，这里只生成信号，不计算收益。

把“信号生成”和“回测计算”分开，是一个很好的习惯。否则每个策略文件里都会复制一堆收益、回撤、画图代码，越写越乱。

4. 回测层：统一处理收益、成本和指标

写一个通用回测函数：

import numpy as np
import pandas as pd

def calc_metrics(equity: pd.Series, returns: pd.Series) -> dict:
    total_return = equity.iloc[-1] / equity.iloc[0] - 1
    annual_return = (1 + total_return) ** (252 / len(equity)) - 1
    drawdown = equity / equity.cummax() - 1
    sharpe = 0 if returns.std() == 0 else returns.mean() / returns.std() * np.sqrt(252)

    return {
        "total_return": total_return,
        "annual_return": annual_return,
        "max_drawdown": drawdown.min(),
        "sharpe": sharpe,
    }

def backtest_single_asset(
    df: pd.DataFrame,
    fee: float = 0.0003,
    slippage: float = 0.0002,
) -> tuple[pd.DataFrame, dict]:
    df = df.copy().dropna(subset=["signal"])

    df["position"] = df["signal"].shift(1).fillna(0)
    df["ret"] = df["close"].pct_change().fillna(0)
    df["strategy_ret"] = df["position"] * df["ret"]

    df["turnover"] = df["position"].diff().abs().fillna(0)
    df["cost"] = df["turnover"] * (fee + slippage)
    df["strategy_ret"] = df["strategy_ret"] - df["cost"]

    df["equity"] = (1 + df["strategy_ret"]).cumprod()
    metrics = calc_metrics(df["equity"], df["strategy_ret"])
    return df, metrics

以后你写任何单标的策略，只要生成 signal，就能复用这套回测逻辑。

这就是工作台的意义：把重复劳动变少，把研究动作变标准。

5. 配置层：每次实验都保存参数

不要把所有参数都写死在代码里。

用一个 JSON 配置：

{
  "name": "ma_cross_600519",
  "symbol": "600519.SH",
  "count": 800,
  "adjust": "forward",
  "short_window": 20,
  "long_window": 60,
  "fee": 0.0003,
  "slippage": 0.0002
}

主程序读取配置：

import json
from data_loader import AlphaFeedDataLoader
from strategies.ma_cross import ma_cross_signal
from backtest import backtest_single_asset

with open("configs/ma_cross.json", "r", encoding="utf-8") as f:
    config = json.load(f)

loader = AlphaFeedDataLoader()
df = loader.load_daily_bars(
    config["symbol"],
    count=config["count"],
    adjust=config["adjust"],
)

df = ma_cross_signal(
    df,
    short_window=config["short_window"],
    long_window=config["long_window"],
)

result, metrics = backtest_single_asset(
    df,
    fee=config["fee"],
    slippage=config["slippage"],
)

print(metrics)

这样你每次实验的输入都能被记录下来。

当你以后回看结果时，不会只剩下一张不知道怎么来的图。

6. 报告层：保存结果，而不是只打印

每次回测建议保存三类东西：

1. 配置。
2. 指标。
3. 净值曲线和明细。

from pathlib import Path
import json
import matplotlib.pyplot as plt

run_dir = Path("reports/runs/ma_cross_600519")
run_dir.mkdir(parents=True, exist_ok=True)

with open(run_dir / "config.json", "w", encoding="utf-8") as f:
    json.dump(config, f, ensure_ascii=False, indent=2)

with open(run_dir / "metrics.json", "w", encoding="utf-8") as f:
    json.dump(metrics, f, ensure_ascii=False, indent=2)

result.to_csv(run_dir / "result.csv", index=False)

result.set_index("trade_date")["equity"].plot(figsize=(10, 5))
plt.tight_layout()
plt.savefig(run_dir / "equity.png", dpi=160)

一个实验目录长这样：

reports/runs/ma_cross_600519/
  config.json
  metrics.json
  result.csv
  equity.png

这就非常舒服了。你可以随时回看，也可以把结果发给 Codex 继续分析。

7. Codex 在工作台里能做什么

当你的项目结构稳定之后，Codex 的作用会被放大。

你可以让它做这些事：

请在 strategies/ 下新增一个 rsi_reversal.py。
要求生成 signal 字段：
1. RSI 小于 30 时买入。
2. RSI 大于 50 时卖出。
3. 不要在策略函数里计算收益。
4. 保持和 ma_cross.py 一样的函数风格。

或者：

请给 backtest_single_asset 增加一个 benchmark 参数。
如果传入 benchmark 收益序列，请输出超额收益和信息比率。

或者：

请检查整个回测代码是否有未来函数。
重点检查 signal、position、ret 的日期关系。

最关键的是：你不再让 Codex 面对一团零散脚本，而是让它在一个清晰项目里工作。它会更容易理解你的意图，也更容易做出稳定修改。

8. 一个适合发给 Codex 的项目提示词

你可以把这段作为工作台的长期说明：

这是我的个人量化研究工作台。

数据源使用 AlphaFeed：
- AlphaFeed 官网：https://alphafeed.org/
- Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart

常用接口：
- 日 K：af.klines.get(symbol, period="1d", count=800, adjust="forward", to_dataframe=True)
- 批量 K 线：af.klines.batch(symbols, period="1d", count=800, adjust="forward", to_dataframe=True)
- 实时行情：af.quotes.get(symbols=symbols, to_dataframe=True)
- 全量 A 股：af.quotes.get(universes="CN_Stock", to_dataframe=True)
- 全量 ETF：af.quotes.get(universes="CN_ETF", to_dataframe=True)
- 五档盘口：af.depth.get(symbol)

项目约定：
1. data_loader.py 只负责取数和缓存。
2. strategies/ 里的策略函数只生成 signal，不计算收益。
3. backtest.py 负责 position、returns、cost、equity、metrics。
4. 所有 signal 都必须 shift 后才能变成 position。
5. 每次实验都要保存 config、metrics、result.csv、equity.png。
6. 不要把 API key 写进代码，使用 ALPHAFEED_API_KEY 环境变量。

有了这段上下文，你以后新增策略会非常顺手。

9. 工作台的下一步：从脚本到小应用

当命令行版本稳定后，可以继续升级：

方向	做法
参数面板	用 Streamlit 做输入表单
图表展示	展示净值、回撤、持仓
实验列表	读取reports/runs 展示历史回测
盘中监控	接入 AlphaFeed 实时行情
自动总结	让 AI 根据 metrics 写研究摘要

但不要一开始就做复杂系统。

个人量化工作台最重要的是“研究闭环”：

提出想法 -> 获取数据 -> 生成信号 -> 回测 -> 保存结果 -> 复盘 -> 继续迭代

只要这个闭环跑通，你的研究效率就会明显提升。

结语

量化研究不是只写一个策略文件。

它更像一套长期工作流：数据稳定、代码清晰、结果可复现、实验可追踪。AlphaFeed 解决数据接入和行情统一的问题，Codex 解决代码生成、调试和重构的问题，而你负责提出问题、判断逻辑、理解风险。

这三者配合起来，普通人也可以搭出一个像样的量化研究环境。

先从一个轻量工作台开始。不要追求完美，追求每次研究都能留下清楚痕迹。

相关链接：

• AlphaFeed 官网：https://alphafeed.org/
• Python SDK 快速开始：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart