年化308%，回撤12%，夏普4.39的季度最强策略，python代码

原创内容第982篇，专注AGI+，AI量化投资、个人成长与财富自由。

策略地址:

http://www.ailabx.com/strategy/68a3633d8c45f7483b1aa0e4

A股全市场数据加载，做了一些选型。

之前咱们做过对比：沪深300成份股，300支600个csv文件，polars加载仅需0.45秒，附python脚本

之前是对比pandas，duckdb和polars加载csv，parquet等。

遇到一些问题，当csv文件多了之后，可能有一些表结构不对，或者数据格式问题，多个dataframe就合不到一块，而且出错后还不容易排查。

今天重新启用hdf5。

import akshare as akimport pandas as pdimport osimport timefrom datetime import datetime
# 创建存储目录history_dir = "../data/csi300/history"inc_dir = "../data/csi300/inc"data_dir = '../data'os.makedirs(history_dir, exist_ok=True)os.makedirs(inc_dir, exist_ok=True)

def get_csi300_constituents():    """获取沪深300成分股"""    df = ak.index_stock_cons_sina(symbol="000300")    # 统一格式：只保留股票代码（去掉市场前缀）    df["code"] = df["code"].str.replace(r"sh\.|sz\.", "", regex=True)    return df["code"].tolist()

def download_adjusted_data(symbol):    """下载个股后复权数据并统一字段"""    try:        # 尝试两种市场代码格式        for prefix in ["sh", "sz"]:            full_code = f"{symbol}"            df = ak.stock_zh_a_hist(                symbol=full_code,                period="daily",                adjust="hfq"  # 后复权            )            if not df.empty:                break        if df.empty:            print(f"⚠️ 未找到数据: {symbol}")            return None
        # 统一字段名称        df = df.rename(columns={            "日期": "date",            "开盘": "open",            "最高": "high",            "最低": "low",            "收盘": "close",            "成交量": "volume"        })        # 添加股票代码列        df.insert(0, "symbol", symbol)        return df[["symbol", "date", "open", "high", "low", "close", "volume"]]
    except Exception as e:        print(f"❌ 下载{symbol}失败: {str(e)}")        return None



    

def split_and_save_data(df, symbol):    """按日期分割数据并保存到不同目录"""    if df is None:        return
    # 转换日期格式    df["date"] = pd.to_datetime(df["date"])
    # 分割数据    history_data = df[df["date"] <= "2024-12-31"]    inc_data = df[df["date"] >= "2025-01-01"]
    # 保存历史数据    if not history_data.empty:        history_path = os.path.join(history_dir, f"{symbol}.csv")        history_data.to_hdf(os.path.join(data_dir,"history.h5"), key='history',                            complib="blosc",  # 使用 blosc 压缩算法                            complevel=9,  # 压缩级别（1-9，越高压缩率越大）                            mode="a", format="table")
        #history_data.to_csv(history_path, index=False, date_format="%Y-%m-%d")
    # 保存增量数据    if not inc_data.empty:
        inc_path = os.path.join(inc_dir, f"{symbol}.csv")        inc_data.to_hdf(os.path.join(data_dir,"inc.h5"), key=symbol,                        complib="blosc",  # 使用 blosc 压缩算法                        complevel=9,  # 压缩级别（1-9，越高压缩率越大）                        mode="a", format="table"                        )        #inc_data.to_csv(inc_path, index=False, date_format="%Y-%m-%d")

# 主程序if __name__ == "__main__":    print("开始获取沪深300成分股...")    symbols = get_csi300_constituents()    print(f"获取到{len(symbols)}只成分股")
    total = len(symbols)    for i, symbol in enumerate(symbols, 1):        print(f"\n下载进度 ({i}/{total}): {symbol}")        start_time = time.time()
        # 下载数据        df = download_adjusted_data(symbol)
        # 处理并保存数据        if df is not None:            split_and_save_data(df, symbol)            time_cost = time.time() - start_time            print(f"✅ 完成 {symbol} | 耗时: {time_cost:.2f}秒 | 记录数: {len(df)}")        else:            print(f"⏭ 跳过 {symbol}")
        # 请求间隔防止被封        time.sleep(1)
    print("\n全部任务完成！")    print(f"历史数据保存至: {os.path.abspath(history_dir)}")    print(f"增量数据保存至: {os.path.abspath(inc_dir)}")