Yfinance股票数据CSV增量更新实现及CSV与SQLite选型咨询

一、实现CSV增量更新（避免覆盖原有数据） #

原代码每次下载全量数据覆盖CSV，可通过**「首次全量下载+后续增量追加」**逻辑解决，核心是判断文件是否存在，按需下载数据：

优化后的完整代码 #

import yfinance as yf
import os
import pandas as pd
from datetime import datetime, timedelta

# 配置固定路径
SYMBOLS_FILE = r'C:\zPythonFilesDir\yfinance\symbolstest.csv'
TARGET_DIR = r'C:\zPythonFilesDir\yfinance\newdatasetstest'

# 自动创建目标目录
os.makedirs(TARGET_DIR, exist_ok=True)

# 读取股票代码列表
with open(SYMBOLS_FILE) as f:
    symbols = [line.strip() for line in f if line.strip()]

for symbol in symbols:
    csv_path = os.path.join(TARGET_DIR, f"{symbol}.csv")
    print(f"处理股票：{symbol}")

    if not os.path.exists(csv_path):
        # 首次运行：下载全量历史数据
        data = yf.download(symbol, start="2023-07-01", end="2023-07-30")
        # 直接添加自定义计算列
        data['new_val'] = 2 * round(data['High'], 2)
        # 保存时保留日期索引
        data.to_csv(csv_path, index=True)
        print(f"首次全量数据已保存：{csv_path}")
    else:
        # 增量更新：读取现有数据的最新日期
        existing_df = pd.read_csv(csv_path, index_col=0, parse_dates=True)
        last_record_date = existing_df.index.max()
        # 从最新日期次日开始下载（yfinance自动跳过非交易日）
        start_date = last_record_date + timedelta(days=1)
        end_date = datetime.today().date()

        # 拉取增量数据
        incremental_data = yf.download(symbol, start=start_date, end=end_date)
        if not incremental_data.empty:
            # 给增量数据添加自定义列
            incremental_data['new_val'] = 2 * round(incremental_data['High'], 2)
            # 追加到CSV，不重复写入表头
            incremental_data.to_csv(csv_path, mode='a', header=False)
            print(f"已追加 {len(incremental_data)} 条新数据")
        else:
            print(f"无新数据可更新（最新记录日期：{last_record_date.date()}）")

关键逻辑说明 #

• 用os.path.exists判断文件是否存在，区分首次全量/增量更新
• 追加模式用mode='a'+header=False，避免重复写入表头
• 解析现有数据的日期索引，确保增量下载的起始日期准确
• yfinance自动跳过周末/节假日，无需额外处理非交易日

二、优化现有代码逻辑 #

原代码分「存储原始数据→读取加自定义列」两步，可整合为一步减少IO操作，同时简化读取逻辑：

优化后的读取代码 #

import os
import pandas as pd
from datetime import datetime

TARGET_DIR = r'C:\zPythonFilesDir\yfinance\newdatasetstest'

# 封装单只股票数据读取函数
def load_stock_data(symbol, filter_recent_days=None):
    csv_path = os.path.join(TARGET_DIR, f"{symbol}.csv")
    if not os.path.exists(csv_path):
        print(f"未找到股票 {symbol} 的数据文件")
        return None
    # 读取时解析日期为索引
    df = pd.read_csv(csv_path, index_col=0, parse_dates=True)
    # 可选：筛选最近N天数据
    if filter_recent_days:
        df = df[df.index >= datetime.today() - timedelta(days=filter_recent_days)]
    return df

# 示例：读取AAPL最近30天数据
aapl_df = load_stock_data("AAPL", filter_recent_days=30)
if aapl_df is not None:
    print("\nAAPL最近30天数据：")
    print(aapl_df.tail())

# 遍历所有股票的最新一条记录
for filename in os.listdir(TARGET_DIR):
    if filename.endswith(".csv"):
        symbol = filename.split(".")[0]
        df = pd.read_csv(os.path.join(TARGET_DIR, filename), index_col=0, parse_dates=True)
        print(f"\n{symbol} 最新记录：")
        print(df.tail(1))

优化点 #

• 下载后直接添加自定义列，无需中间目录存储原始数据
• 封装读取函数，支持按股票代码和时间范围筛选数据
• 读取时解析日期为索引，方便后续时间维度的操作

三、CSV+Pandas vs SQLite的优劣与最佳实践 #

1. CSV+Pandas 优缺点 #

优点

• 简单直观：纯文本格式，可直接用Excel/文本编辑器打开查看
• 无额外依赖：Pandas原生支持，无需安装数据库驱动
• 轻量便携：单文件存储，易于备份和传输
• 适合小数据量：数据量<10万条时，读写速度足够

缺点

• 增量更新繁琐：需手动处理表头去重、日期校验
• 查询效率低：必须加载全量数据到内存才能筛选/聚合
• 数据一致性差：无事务支持，写入中断易导致数据损坏
• 不支持复杂操作：无法关联多数据集、执行SQL查询

2. SQLite 优缺点 #

优点

• 增量更新高效：支持INSERT语句，可通过主键约束自动去重
• 查询灵活：支持完整SQL语法，可快速筛选、聚合、关联数据
• 数据安全：支持事务，写入失败可回滚，避免数据损坏
• 适合大数据量：无需加载全量数据到内存，内存占用低
• 单文件存储：和CSV一样，数据库文件可直接备份传输

缺点

• 有学习成本：需要掌握基础SQL语法
• 无法直接编辑：二进制格式，需用SQL工具或代码操作
• 小数据量下读写速度略低于CSV（差异可忽略）

3. 最佳实践 #

• 小数据量、仅简单读写：用CSV+Pandas，开发成本最低
• 大数据量、需频繁增量更新/复杂查询：用SQLite+Pandas（Pandas支持直接读写SQLite）
• 混合方案：用SQLite存储原始数据，需要导出报表时再转存为CSV

SQLite实现示例（简化版） #

import yfinance as yf
import pandas as pd
import sqlite3
from datetime import datetime, timedelta

DB_PATH = r'C:\zPythonFilesDir\yfinance\stock_data.db'

# 连接SQLite数据库（不存在则自动创建）
conn = sqlite3.connect(DB_PATH)

# 读取股票代码
with open(SYMBOLS_FILE) as f:
    symbols = [line.strip() for line in f if line.strip()]

for symbol in symbols:
    # 查询现有数据的最新日期
    try:
        last_date = pd.read_sql(f"SELECT MAX(Date) FROM {symbol}", conn).iloc[0,0]
    except:
        last_date = "2023-06-30"  # 无数据时默认起始日期

    # 下载增量数据
    incremental_data = yf.download(symbol, start=last_date, end=datetime.today().date())
    if not incremental_data.empty:
        incremental_data['new_val'] = 2 * round(incremental_data['High'], 2)
        # 写入数据库，自动创建表（若不存在），追加数据
        incremental_data.to_sql(symbol, conn, if_exists='append', index=True, index_label='Date')
        print(f"已追加 {len(incremental_data)} 条数据到{symbol}表")

# 关闭数据库连接
conn.close()

# 读取数据示例
conn = sqlite3.connect(DB_PATH)
aapl_df = pd.read_sql("SELECT * FROM AAPL WHERE Date >= '2023-07-01'", conn)
conn.close()
print(aapl_df.tail())

内容的提问来源于stack exchange，提问作者andypappy