我用StockApi实现双均线趋势跟踪：个股版 Python 实现，均线直接调指标接口（附完整代码）

import os
import requests
import pandas as pd

TOKEN = os.getenv("STOCKAPI_TOKEN", "你的Token")
BASE = "https://www.stockapi.com.cn/v1"

def call(path, **params):
    """StockApi 统一调用：code==20000 为成功"""
    params["token"] = TOKEN
    r = requests.get(f"{BASE}{path}", params=params, timeout=15)
    r.raise_for_status()
    d = r.json()
    if d.get("code") == 20000:
        return d["data"]
    raise RuntimeError(f"{path} 调用失败: {d.get('msg')}")

def to_df(data):
    """兼容三种返回结构：list[dict] / 列式字典 / 分页 records"""
    if isinstance(data, list):
        df = pd.DataFrame(data)
    elif isinstance(data, dict):
        if "records" in data:
            df = pd.DataFrame(data["records"])
        else:  # K线接口是 {字段: [值,...]} 的列式返回
            df = pd.DataFrame({k: v for k, v in data.items() if isinstance(v, list)})
    else:
        raise TypeError("未知返回结构")
    if "date" in df.columns:
        df["date"] = pd.to_datetime(df["date"])
        df = df.sort_values("date").reset_index(drop=True)
    return df

def get_kline(code, start, end, cycle=100):
    """前复权日K（100=日 101=周 102=月），列式返回，含 PE/PB/市值/换手"""
    return to_df(call("/base/day", code=code, startDate=start,
                      endDate=end, calculationCycle=cycle))

def get_ma(code, start, end, periods="20,60,200"):
    """MA 指标接口：一次拿多条均线"""
    return to_df(call("/quota/ma2", code=code, ma=periods,
                      startDate=start, endDate=end, calculationCycle=100))

def align(px, ind):
    """K线无 date 字段时按交易日顺序尾部对齐；有 date 则按日期合并"""
    if "date" in px.columns and "date" in ind.columns:
        return pd.merge(px, ind, on="date", how="inner").sort_values("date").reset_index(drop=True)
    n = min(len(px), len(ind))
    left = px.tail(n).reset_index(drop=True)
    right = ind.tail(n).drop(columns=[c for c in ("date", "api_code") if c in ind.columns]).reset_index(drop=True)
    return pd.concat([left, right], axis=1)

def backtest_dual_ma(code, start, end):
    px = get_kline(code, start, end)
    ma = get_ma(code, start, end)
    df = align(px, ma)
    for c in ["ma20", "ma60", "ma200"]:
        df[c] = pd.to_numeric(df[c], errors="coerce")
    df["close"] = pd.to_numeric(df["close"], errors="coerce")
    df = df.dropna(subset=["ma200"]).reset_index(drop=True)

    pos, hold = [], False
    for i in range(1, len(df)):
        r, p = df.iloc[i], df.iloc[i - 1]
        if not hold and p["ma20"] <= p["ma60"] and r["ma20"] > r["ma60"] and r["close"] > r["ma200"]:
            hold = True
        elif hold and (r["ma20"] < r["ma60"] or r["close"] < r["ma200"]):
            hold = False
        pos.append(1 if hold else 0)
    df["pos"] = [0] + pos
    df["ret"] = df["close"].pct_change().fillna(0)
    df["strategy_ret"] = df["pos"].shift(1).fillna(0) * df["ret"]
    df["equity"] = (1 + df["strategy_ret"]).cumprod()

    ann = df["strategy_ret"].mean() * 243
    dd = (df["equity"] / df["equity"].cummax() - 1).min()
    print(f"{code} 年化 {ann:.1%} 最大回撤 {dd:.1%} 持仓天数占比 {df['pos'].mean():.1%}")
    return df

# 单只回测
backtest_dual_ma("600519", "2018-01-01", "2026-09-30")

# 批量：先拿全市场列表，再按市值初筛后逐只跑（注意 40 次/秒的频率上限）
stocks = call("/base/all")
print("全市场股票数:", len(stocks), "| 示例字段:", stocks[0])