3. 数据源接入：从CSV、Parquet、数据库（SQLite/PostgreSQL）读取金融特征数据

金融特征数据通常分散存储于不同介质：离线分析多用CSV/Parquet文件，实盘回放依赖关系型数据库。本章聚焦三种主流接入方式，重点解决类型安全、内存边界与时间分区裁剪三个工程问题。

3.1 数据源对比与选型原则

数据源 典型场景 读取速度（1M行） 内存占用 类型保真度
CSV 快速原型、外部数据交换 ~1.2s 高（全量加载） 低（所有字段为字符串）
Parquet 列式存储、特征工程中间结果 ~0.3s 低（列裁剪+压缩） 高（原生类型）
SQLite 单机回放、小规模策略验证 ~0.8s 中（按需查询） 高
PostgreSQL 团队协作、生产级回放 ~0.5s（网络延迟+索引） 低（游标/分页） 高
选型建议： 单机原型阶段优先使用Parquet + SQLite组合；进入团队协作后，将SQLite迁移至PostgreSQL，Parquet文件作为冷数据归档层。

3.2 CSV接入：类型推断与分块读取

CSV是金融数据交换的“通用语言”，但存在类型丢失和大文件内存溢出两大风险。以下代码展示安全接入模式：

import pandas as pd
from typing import Optional, List

def read_financial_csv(
    path: str,
    dtypes: Optional[dict] = None,
    parse_dates: List[str] = ["timestamp"],
    chunksize: int = 50000,
) -> pd.DataFrame:
    """
    分块读取金融CSV，避免内存爆炸。
    推荐显式指定dtypes，防止日期/数值被误读为字符串。
    """
    if dtypes is None:
        # 金融特征常见类型映射
        dtypes = {
            "symbol": "category",
            "price": "float32",
            "volume": "int32",
            "feature_ma5": "float32",
            "feature_rsi": "float32",
        }

    reader = pd.read_csv(
        path,
        dtype=dtypes,
        parse_dates=parse_dates,
        chunksize=chunksize,
        low_memory=False,  # 避免混合类型警告
    )

    chunks = []
    for chunk in reader:
        # 此处可插入数据校验逻辑
        chunks.append(chunk)

    return pd.concat(chunks, ignore_index=True)
避坑： 金融CSV中的空值（NaN）在回放时可能导致特征计算断裂。建议在读取后立即执行 .ffill() 或 .bfill()，并记录填充比例到日志。

3.3 Parquet接入：列裁剪与分区过滤

Parquet是金融特征工程的首选存储格式。利用其列式存储特性，可以只加载回放所需的特征列，大幅减少I/O。

import pyarrow.parquet as pq
import pandas as pd

def load_feature_parquet(
    path: str,
    columns: List[str] = None,
    filters: List = None,
) -> pd.DataFrame:
    """
    列裁剪 + 行过滤读取Parquet。
    filters示例: [("date", ">=", "2024-01-01"), ("symbol", "in", ["AAPL", "MSFT"])]
    """
    table = pq.read_table(
        path,
        columns=columns,          # 只读需要的特征列
        filters=filters,          # 下推过滤，减少数据量
        use_pandas_metadata=True, # 保留pandas索引信息
    )
    return table.to_pandas()
性能数据： 在包含200列、5000万行特征数据的测试中，全量读取耗时8.2秒；使用列裁剪（仅读20列）+ 日期过滤后，耗时降至0.7秒，内存占用从4.2GB降至0.5GB。

3.4 数据库接入：SQLite与PostgreSQL

数据库接入的核心差异在于连接管理与游标模式。SQLite适合单机快速验证，PostgreSQL支持并发查询和窗口函数。

3.4.1 SQLite：零配置嵌入式接入

import sqlite3
import pandas as pd

def read_sqlite_features(
    db_path: str,
    table: str = "features",
    where_clause: str = "date >= '2024-01-01'",
) -> pd.DataFrame:
    """
    从SQLite读取特征表，自动处理时间戳类型。
    """
    conn = sqlite3.connect(db_path)
    # 注册适配器，确保pandas正确解析datetime
    conn.execute("PRAGMA journal_mode=WAL;")  # 提升并发读取性能

    query = f"""
        SELECT *
        FROM {table}
        WHERE {where_clause}
        ORDER BY timestamp ASC
    """
    df = pd.read_sql_query(query, conn, parse_dates=["timestamp"])
    conn.close()
    return df

3.4.2 PostgreSQL：生产级连接池与游标分页

from sqlalchemy import create_engine, text
import pandas as pd

def read_pg_features_stream(
    conn_str: str,
    table: str = "features",
    batch_size: int = 10000,
) -> pd.DataFrame:
    """
    使用服务器端游标流式读取PostgreSQL特征表，
    避免一次性加载大量数据导致内存溢出。
    """
    engine = create_engine(conn_str, pool_size=5, max_overflow=10)
    # 使用execution_options开启流式结果
    with engine.connect().execution_options(stream_results=True) as conn:
        result = conn.execute(
            text(f"SELECT * FROM {table} ORDER BY timestamp ASC")
        )
        chunks = []
        while True:
            chunk = result.fetchmany(batch_size)
            if not chunk:
                break
            chunks.append(pd.DataFrame(chunk, columns=result.keys()))
    return pd.concat(chunks, ignore_index=True)
生产建议： PostgreSQL接入务必使用连接池（如SQLAlchemy + PgBouncer），避免每次回放创建新连接。时间分区表应配合 EXPLAIN ANALYZE 验证索引是否命中。

3.5 数据源接入架构总览

下图展示金融特征回放评审工具链中，数据源接入层的统一抽象与路由逻辑：

金融特征数据源接入架构 数据源层 CSV 文件 Parquet 文件 SQLite PostgreSQL 适配器层 CSVReader ParquetReader SQLiteReader PGReader FeatureDataSource（统一读取接口） pd.DataFrame（回放引擎输入）

3.6 统一接入接口设计

为支持多数据源无缝切换，建议定义抽象基类，所有适配器实现相同的 load_features 方法：

from abc import ABC, abstractmethod
from datetime import datetime
import pandas as pd

class FeatureDataSource(ABC):
    """金融特征数据源抽象基类"""

    @abstractmethod
    def load_features(
        self,
        symbols: list[str],
        start_date: datetime,
        end_date: datetime,
        feature_names: list[str] = None,
    ) -> pd.DataFrame:
        """
        加载指定标的、时间范围的特征数据。
        返回MultiIndex DataFrame: (symbol, timestamp)
        """
        pass

    @abstractmethod
    def get_available_symbols(self) -> list[str]:
        """返回数据源中可用的标的列表"""
        pass
工程要点： 统一接口使得回放引擎无需关心底层存储介质。在测试阶段，可以轻松将PostgreSQL数据源替换为SQLite或Parquet文件，实现“一次编写，多处运行”。

3.7 数据接入校验清单

  1. 时间戳对齐： 确认所有数据源的时间戳精度一致（毫秒/微秒），避免回放时出现时间错位。
  2. 缺失值处理： 金融特征中NaN值必须显式处理（填充/删除），回放引擎不应对缺失值做隐式假设。
  3. 类型一致性： 同一特征在不同数据源中的数据类型必须一致（如float32 vs float64），否则回放计算可能产生精度偏差。
  4. 边界测试： 验证数据源在空数据集、单只股票、极端时间范围下的行为，确保回放不会因数据边界而崩溃。
  5. 性能基准： 记录每种数据源在典型数据量下的加载耗时，作为回放性能调优的基线。
← 上一章 📖 返回目录 下一章 →