第 41 篇：工程化改造项目结构与 DuckDB

quant-project/
├── config/
│   └── strategy.yaml        # 策略参数：周期、费率、区间
├── data/
│   └── raw/ parquet 文件    # 本地行情库
├── strategies/
│   └── ma_cross.py          # 策略：只管信号
├── backtest/
│   └── engine.py            # 引擎：只管撮合与记账
├── scripts/
│   └── update_data.py       # 数据更新入口
├── logs/                    # 运行日志（第 43 篇展开）
└── main.py                  # 总入口

# ---------- 0. 准备：造一份行情 Parquet（真实项目由 scripts/update_data.py 生成） ----------
import numpy as np
import pandas as pd
from pathlib import Path

Path("data/raw").mkdir(parents=True, exist_ok=True)
rng = np.random.default_rng(41)
n = 1000
df = pd.DataFrame({
    "code": "600519",
    "date": pd.bdate_range("2021-01-01", periods=n),
    "close": 100 * np.exp(np.cumsum(rng.normal(0.0004, 0.015, n))),
})
df["open"] = df["close"] * (1 + rng.normal(0, 0.003, n).cumsum() * 0)
df.to_parquet("data/raw/600519.parquet", index=False)
print(f"已生成行情文件: {len(df)} 行")

# ---------- config/strategy.yaml 的内容 ----------
# 双均线策略参数（示意）
symbol: "600519"
fast: 5
slow: 20
fee: 0.001
start: "2021-01-01"
end: "2024-12-31"

# ---------- 1. strategies/ma_cross.py 的内容 ----------
import pandas as pd

def generate_signals(df, fast, slow):
    """输入行情 DataFrame，输出带 signal 列的 DataFrame。只管信号。"""
    out = df.copy()
    out["ma_fast"] = out["close"].rolling(fast).mean()
    out["ma_slow"] = out["close"].rolling(slow).mean()
    out["signal"] = (out["ma_fast"] > out["ma_slow"]).astype(int)
    return out

# ---------- 2. backtest/engine.py 的内容 ----------
import pandas as pd

def run_vectorized(df, fee):
    """输入带 signal 列的 DataFrame，输出策略日收益。只管执行。"""
    pos = df["signal"].shift(1).fillna(0)      # 信号次日生效，无未来函数
    ret = df["close"].pct_change().fillna(0)
    turnover = pos.diff().abs().fillna(0)
    return pos * ret - turnover * fee

# ---------- 3. main.py：读配置 → DuckDB 取数 → 跑策略 ----------
import duckdb
import yaml
import pandas as pd
from pathlib import Path
from strategies.ma_cross import generate_signals
from backtest.engine import run_vectorized

def load_config(path="config/strategy.yaml"):
    with open(path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)

def query_data(cfg):
    """DuckDB 直接查 Parquet，WHERE 在 SQL 里做，少拉无效数据"""
    sql = f"""
        SELECT date, open, close
        FROM 'data/raw/{cfg['symbol']}.parquet'
        WHERE date BETWEEN '{cfg['start']}' AND '{cfg['end']}'
        ORDER BY date
    """
    df = duckdb.sql(sql).df()                   # 查询结果 → DataFrame
    df["date"] = pd.to_datetime(df["date"])
    return df.set_index("date")

def main():
    cfg = load_config()
    df = query_data(cfg)
    sig = generate_signals(df, cfg["fast"], cfg["slow"])
    strat = run_vectorized(sig, cfg["fee"])
    nav = (1 + strat).cumprod()
    print(f"参数: fast={cfg['fast']}, slow={cfg['slow']}, "
          f"fee={cfg['fee']}")
    print(f"回测区间: {df.index[0].date()} ~ {df.index[-1].date()}, "
          f"{len(df)} 根K线")
    print(f"期末净值(示意): {nav.iloc[-1]:.3f}")
    return cfg, nav

cfg, nav = main()

# ---------- 4. DuckDB 的分析威力：注册 DataFrame 直接 SQL ----------
import duckdb

# DataFrame 注册成临时表视图，SQL 与 pandas 无缝互通
duckdb.register("nav_view", nav.reset_index().rename(
    columns={"index": "date", 0: "nav"}))
stats = duckdb.sql("""
    SELECT
        date_trunc('month', date) AS 月份,
        last_value(nav) OVER (
            PARTITION BY date_trunc('month', date)
            ORDER BY date
            ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
        ) AS 月末净值
    FROM nav_view
    QUALIFY row_number() OVER (
        PARTITION BY date_trunc('month', date) ORDER BY date DESC
    ) = 1
    ORDER BY 月份
""").df()
print(stats.head(6).round(3).to_string(index=False))

示例输出（示意，模拟行情结果）:
已生成行情文件: 1000 行
参数: fast=5, slow=20, fee=0.001
回测区间: 2021-01-01 ~ 2024-12-06, 1000 根K线
期末净值(示意): 1.412
     月份  月末净值
2021-01-01    1.018
2021-02-01    0.993
2021-03-01    1.055
2021-04-01    1.021
2021-05-01    1.090
2021-06-01    1.112