获取沪市主板A股所有股票20250430至20260430的日线行情数据，包括开盘价、收盘价、最高价、最低价和成交量。下面直接上代码。

import akquant as aq
import akshare as ak
import pandas as pd
from akquant import Bar, Strategy

def get_sh_mainboard_stocks():
    """
    精准获取：沪市主板A股列表
    排除科创板，纯主板股票
    """
    # 1. 获取全市场A股代码名称（akshare永久稳定接口）
    stock_all = ak.stock_info_a_code_name()

    # 2. 严格筛选 沪市主板（只保留600/601/603/605开头）
    sh_mainboard_pattern = r"^(600|601|603|605)"
    stock_sh_main = stock_all[stock_all["code"].str.match(sh_mainboard_pattern)].copy()

    # 3. 格式化列名 + 拼接交易代码
    stock_sh_main["交易代码"] = "sh" + stock_sh_main["code"]  # 适配akshare数据获取
    stock_sh_main.columns = ["股票代码", "股票名称", "交易代码"]

    return stock_sh_main.reset_index(drop=True)


def get_data(gupiao) -> pd.DataFrame:
    """
    步骤 1: 数据获取.

    使用 akshare 获取浦发银行 (600000) 的历史日线数据。
    """
    print("正在获取数据...")
    # 获取前复权数据
    df1 = ak.stock_zh_a_daily(
        symbol=gupiao, start_date="20240430", end_date="20250430", adjust="qfq"
    )
    df1["symbol"] = gupiao.replace("sh", "")
    if "date" not in df1.columns:
        df1 = df1.reset_index().rename(columns={"index": "date"})
    return df1  # type: ignore


class DualMAStrategy(Strategy):
    """
    步骤 2: 策略定义.

    双均线策略实现。
    """

    def __init__(self) -> None:
        """策略初始化."""
        super().__init__()
        self.short_window = 5
        self.long_window = 20
        self.warmup_period = self.long_window

    def on_start(self) -> None:
        """策略初始化时调用."""
        print("策略初始化...")

    def on_bar(self, bar: Bar) -> None:
        """核心交易逻辑."""
        symbol = bar.symbol

        # 1. 获取历史数据
        closes = self.get_history(count=self.long_window, symbol=symbol, field="close")

        # 2. 计算均线
        ma5_curr = closes[-self.short_window :].mean()
        ma20_curr = closes[-self.long_window :].mean()

        # 3. 获取持仓
        position = self.get_position(symbol)

        # 4. 交易信号
        if ma5_curr > ma20_curr and position == 0:
            print(
                f"[{bar.timestamp_str}] 金叉买入 (MA5={ma5_curr:.2f}, "
                f"MA20={ma20_curr:.2f})"
            )
            self.order_target_percent(0.95, symbol)

        elif ma5_curr < ma20_curr and position > 0:
            print(
                f"[{bar.timestamp_str}] 死叉卖出 (MA5={ma5_curr:.2f}, "
                f"MA20={ma20_curr:.2f})"
            )
            self.order_target_percent(0.0, symbol)


if __name__ == "__main__":
    # df = get_sh_mainboard_stocks()
    # 1. 获取沪市主板股票列表
    stock_df = get_sh_mainboard_stocks()
    print(f"共获取沪市主板股票：{len(stock_df)} 只")

    # 2. 创建空列表，存储所有股票的回测指标
    result_list = []

    # 3. 遍历所有股票，批量回测
    for idx, row in stock_df.iterrows():
        trade_code = row["交易代码"]  # sh600000
        stock_code = row["股票代码"]  # 600000
        stock_name = row["股票名称"]  # 浦发银行
        # 1. 准备数据
        df = get_data(trade_code)

        # 2. 运行回测
        print("开始回测...")
        result = aq.run_backtest(
            strategy=DualMAStrategy,
            data=df,
            initial_cash=100_000,
            commission_rate=0.0003,
            stamp_tax_rate=0.001,
            lot_size=100,
        )
        # ✅ 提取你需要的指标（修正你写错的对应关系）
        result_list.append({
            "股票代码": stock_code,
            "股票名称": stock_name,
            "交易代码": trade_code,
            "累计收益率": result.metrics.total_return_pct,
            "最大回撤率": result.metrics.max_drawdown_pct,
            "夏普比率": result.metrics.sharpe_ratio,
            "卡玛比率": result.metrics.calmar_ratio,
            "年化波动率": result.metrics.volatility,  # 你之前写错成calmar_ratio
            "胜率": result.metrics.win_rate  # 你之前写错成calmar_ratio
        })
        # result_data.append({
        #     "股票代码": stock_code,
        #     "股票名称": stock_name,
        #     "累计收益率(%)": round(res.total_return * 100, 2),
        #     "最大回撤率(%)": round(res.max_drawdown * 100, 2),
        #     "夏普比率": round(res.sharpe_ratio, 2),
        #     "卡玛比率": round(res.calmar_ratio, 2),
        #     "年化波动率(%)": round(res.annual_volatility * 100, 2),
        #     "胜率(%)": round(res.win_rate * 100, 2)
        # })
        print("✅ 回测成功")

    # 4. 所有股票回测完成，整理结果
    if len(result_list) == 0:
        print("\n无有效回测结果！")
    else:
        # 转DataFrame
        final_df = pd.DataFrame(result_list)

        # ✅ 按 累计收益率 从高到低排序
        final_df = final_df.sort_values(by="累计收益率", ascending=False).reset_index(drop=True)

        # ✅ 保存为CSV文件
        final_df.to_csv("沪市主板A股双均线回测结果.csv", index=False, encoding="utf-8-sig")

        # 打印预览
        print("\n" + "=" * 50)
        print("回测完成！结果已保存")
        print("=" * 50)
        print(final_df.head(10))

运行结果：