上一篇文章介绍了五福闹新春转换到PTrade(https://mp.weixin.qq.com/s/xpmmPd3ojSuleYbzEXZJLw)，但是PTrade平台无法直接获取到ETF的净值，当时这一参数就没有使用。其实，ETF作为交易所交易基金，其净值数据反映了基金资产的实际价值变化，是分析基金表现、构建投资策略的重要依据。本文详细介绍如何系统性地获取全市场ETF历史净值数据，并构建一个稳健的数据获取与处理框架。

数据获取策略的整体逻辑

本策略采用模块化设计，通过分批次处理、增量更新和完整性校验三个核心机制，确保数据获取的效率和可靠性。整体流程遵循获取-处理-存储-校验的循环模式，每个环节都设置了相应的容错机制和数据验证步骤。

策略的核心逻辑基于以下原则：首先获取全市场ETF基础信息，然后分批获取历史净值数据，在获取过程中进行数据清洗和去重处理，最后将有效数据持久化存储。这种设计既考虑了数据获取的稳定性，也兼顾了处理大规模数据时的性能要求。

代码模块详细解析

主函数结构设计

defget_all_etf_nav_history(start_date: str = "20240101", end_date: str = None, output_file: str = None, batch_size: int = 50):"""    获取全市场所有 ETF 从 2024 年至今的历史净值，保存到一个 CSV 文件中    参数:        start_date: 开始日期，格式 'YYYYMMDD'        end_date: 结束日期，格式 'YYYYMMDD'，默认为当前日期        output_file: 输出 CSV 文件路径        batch_size: 每批获取的 ETF 数量，默认 50    返回:        pd.DataFrame: 包含所有 ETF 历史净值的 DataFrame    """

主函数采用参数化设计，提供灵活的日期范围控制、输出路径配置和批处理大小调整。默认参数设置为从2024年1月1日开始获取数据，批处理大小为50只ETF，这些默认值基于实际测试中对API稳定性和处理效率的平衡考虑。

初始化与参数处理

if end_date isNone:        end_date = datetime.now().strftime("%Y%m%d")if output_file isNone:        output_file = "ai/data/all_etf_nav_2024.csv"# 确保输出目录存在    output_dir = os.path.dirname(output_file)if output_dir andnot os.path.exists(output_dir):        os.makedirs(output_dir)        print(f"创建目录：{output_dir}")

初始化阶段完成三个关键任务：设置默认结束日期为当前日期，定义默认输出文件路径，确保输出目录存在。目录创建逻辑采用惰性创建策略，只有在目录不存在且路径非空时才创建，避免了不必要的目录创建操作。

ETF代码列表获取

# 获取所有 ETF 代码    print("正在获取全市场 ETF 代码列表...")    etf_list = get_all_etf_codes()ifnot etf_list:        print("未获取到任何 ETF 代码")return pd.DataFrame()    print(f"共获取到 {len(etf_list)} 只 ETF")    print(f"将分 {len(etf_list) // batch_size + 1} 批次获取，每批 {batch_size} 只")

ETF代码获取是整个流程的起点。通过调用get_all_etf_codes()函数获取全市场ETF基础信息，包括代码和名称。获取成功后计算总批次数，为后续的分批处理提供规划依据。如果获取失败，函数提前返回空DataFrame，避免后续无效操作。

现有数据加载与唯一键管理

# 加载已存在的数据唯一键    print("\n检查现有数据...")    existing_keys = load_existing_keys(output_file)# 如果文件已存在，先加载完整数据    all_df = pd.DataFrame()if os.path.exists(output_file):try:            all_df = pd.read_csv(output_file, encoding='utf-8-sig')            print(f"已加载 {len(all_df)} 条现有数据")except Exception as e:            print(f"加载现有数据失败：{e}")

增量更新机制的核心在于唯一键管理。通过load_existing_keys()函数从现有数据文件中提取(nav_date, etf_code)组合作为唯一键，存储在集合中供后续去重使用。同时加载现有完整数据到内存，为后续的数据合并做准备。采用utf-8-sig编码确保CSV文件在不同环境下的兼容性。

分批处理主循环

    success_count = 0    fail_count = 0    fail_codes = []# 分批获取数据    print(f"\n开始获取 {start_date} 至 {end_date} 的历史净值数据...")for i in range(0, len(etf_list), batch_size):        batch_etf_list = etf_list[i:i + batch_size]        batch_num = (i // batch_size) + 1        total_batches = (len(etf_list) + batch_size - 1) // batch_size        print(f"\n{'='*60}")        print(f"第 {batch_num}/{total_batches} 批次 (ETF {i+1}-{min(i+batch_size, len(etf_list))})")        print(f"{'='*60}")

分批处理采用固定窗口滑动的方式遍历ETF列表。每个批次处理batch_size只ETF，通过计算批次编号和总批次数为用户提供清晰的进度反馈。进度显示采用分隔线和详细描述，增强可读性。

单只ETF数据处理

        batch_data = []        batch_new_records = 0        batch_skip_records = 0        batch_fail_count = 0for etf_info in tqdm(batch_etf_list, desc=f"批次 {batch_num} 进度"):            etf_code = etf_info['code']            etf_name = etf_info['name']try:                df = get_etf_nav_history(etf_code, start_date, end_date)if df isnotNoneandnot df.empty:# 添加 ETF 代码和名称列                    df['etf_code'] = etf_code                    df['etf_name'] = etf_name# 重排序，把代码和名称放到前面                    cols = df.columns.tolist()                    cols = ['etf_code', 'etf_name'] + [c for c in cols if c notin ['etf_code', 'etf_name']]                    df = df[cols]

单只ETF处理采用try-except结构确保单点故障不影响整体流程。通过get_etf_nav_history()函数获取指定ETF的历史净值数据，成功后添加ETF代码和名称作为标识列。列重排序操作确保输出数据结构的统一性，将标识信息置于前列便于查看。

数据去重与唯一性校验

# 检查重复数据                    new_rows = []for _, row in df.iterrows():                        key = (row['nav_date'], row['etf_code'])if key notin existing_keys:                            new_rows.append(row)                            existing_keys.add(key)                            batch_new_records += 1else:                            batch_skip_records += 1if new_rows:                        batch_data.append(pd.DataFrame(new_rows))                    success_count += 1else:                    fail_count += 1                    batch_fail_count += 1                    fail_codes.append(etf_code)except Exception as e:                fail_count += 1                batch_fail_count += 1                fail_codes.append(etf_code)                print(f"\nETF {etf_code} 获取失败：{e}")

去重机制基于(nav_date, etf_code)联合唯一键。通过遍历每行数据，检查其唯一键是否已存在于existing_keys集合中，实现精确去重。新增记录同时更新唯一键集合，确保内存中的唯一性状态与文件存储一致。失败处理记录失败计数和具体代码，便于后续排查。

批次数据持久化

# 增量写入本批次数据if batch_data:            batch_df = pd.concat(batch_data, ignore_index=True)# 合并到总数据ifnot all_df.empty:                all_df = pd.concat([all_df, batch_df], ignore_index=True)else:                all_df = batch_df# 保存到 CSV（覆盖写入）            save_to_csv(all_df, output_file)            print(f"\n批次 {batch_num} 完成：新增 {batch_new_records} 条，跳过 {batch_skip_records} 条重复数据")            print(f"本批次失败 {batch_fail_count} 只，当前总记录数：{len(all_df)}")else:            print(f"\n批次 {batch_num} 完成：无新增数据，失败 {batch_fail_count} 只")# 每批次完成后稍作停顿        time.sleep(2)

批次数据持久化采用全量覆盖策略。将本批次新增数据合并到内存中的总数据集，然后调用save_to_csv()函数覆盖写入文件。这种方式虽然每次写入都是全量数据，但保证了数据的完整性和一致性。批次间2秒的停顿时间减少了对数据源的请求压力，提高长期运行的稳定性。

最终统计与结果输出

# 最终统计    print(f"\n{'='*60}")    print("数据获取完成！")    print(f"{'='*60}")    print(f"成功获取 {success_count} 只 ETF 的历史数据")    print(f"失败 {fail_count} 只 ETF")if fail_codes:# 保存失败的 ETF 列表        fail_file = output_file.replace('.csv', '_failed.txt')with open(fail_file, 'w', encoding='utf-8') as f:for code in fail_codes:                f.write(code + '\n')        print(f"失败的 ETF 代码已保存到：{fail_file}")        print(f"失败的 ETF 代码（前 20 个）: {fail_codes[:20]}{'...'if len(fail_codes) > 20else''}")ifnot all_df.empty:        print(f"\n数据已保存到：{output_file}")        print(f"总记录数：{len(all_df)}")return all_df

最终统计提供完整的执行结果汇总，包括成功和失败数量。失败ETF代码单独保存到文本文件，便于后续重试或问题排查。结果显示前20个失败代码，在控制信息量和提供足够信息间取得平衡。函数返回完整的DataFrame，支持链式调用和进一步处理。

数据分析功能

defanalyze_etf_data(df: pd.DataFrame):"""    分析 ETF 数据    参数:        df: ETF 历史数据 DataFrame    """if df.empty:        print("数据为空，无法分析")return    print("\n=== ETF 数据统计 ===")    print(f"总记录数：{len(df)}")    print(f"ETF 数量：{df['etf_code'].nunique()}")if'nav_date'in df.columns:# 转换为日期格式        df['nav_date'] = pd.to_datetime(df['nav_date'], errors='coerce')        print(f"日期范围：{df['nav_date'].min()} 至 {df['nav_date'].max()}")# 按 ETF 统计if'nav_unit'in df.columns:        etf_stats = df.groupby('etf_code').agg({'etf_name': 'first','nav_unit': ['count', 'min', 'max', 'last']        }).reset_index()        etf_stats.columns = ['代码', '名称', '记录数', '最低单位净值', '最高单位净值', '最新单位净值']        print("\nETF 记录数统计（前 10）:")        print(etf_stats.sort_values('记录数', ascending=False).head(10))

数据分析函数提供基本的数据质量检查和统计信息。首先检查数据完整性，然后计算关键统计指标：总记录数、唯一ETF数量、日期范围。净值数据统计按ETF分组，计算记录数、净值范围和新净值。错误处理采用coerce模式，将无效日期转换为NaT，避免转换失败导致程序中断。

ETF 数据获取完成，执行结果：

成功获取 1457 只 ETF 的历史净值数据 失败 0 只 ETF 总记录数：599,955 条 日期范围：2024-01-02 至 2026-04-10 耗时：约 43 分钟 数据已保存到：ai/data/all_etf_nav_2024.csv 数据统计（前 10 只 ETF 按记录数）：

代码 名称 记录数 最低单位净值 最高单位净值 最新单位净值512200 房地产 ETF 上海 549 0.4094 1.8208 1.3466159968 深证 500ETF 深时 549 5.3378 11.0177 10.1717159971 创业板 ETF 深证 549 1.0539 2.3079 1.1972159972 5 年期国债 ETF 深证 549 110.0922 118.3482 118.3482159973 深证 300ETF 深证远望 549 0.9693 1.9459 1.9349

程序执行完毕，所有数据已完整保存到 CSV 文件中。回复”净值”，可获取完整数据。

PS: 源码下载，请移步知识星球！

星球中整理了文章中涉及到的源码，加入后即可以看到之前发的源码，三天内不满意可以退款。也会将收集到的机构研报，优秀策略源码，群友常咨询的问题整理到不同标签下，做为知识库沉淀下来。

所有策略仅用于学习和研究，不保证交易收益，不作为投资建议，风险自负，所有收益仅表示历史回测收益，不表示未来收益。大QMT和ptrade请充分使用模拟盘测试，miniQMT使用模拟账号测试。