数据源接入与ETL：API数据采集、数据库连接、CSV/Excel批量导入、增量更新策略

数据源接入，说白了就是解决「数据从哪来」的问题。我做了这么多年金融数据处理，见过太多团队在第一步就栽跟头——API限流没处理好、数据库连接池爆了、CSV编码搞错导致乱码……这些坑我基本都踩过一遍。今天咱们就把这些环节掰开揉碎，讲清楚。

API数据采集：别让接口把你卡住

金融数据API有个特点——量大、频次高、动不动就限流。我刚开始对接某交易所行情接口时，没做限流控制，结果对方直接把我IP封了24小时。嗯，从那以后我再也不敢小看API调用的优雅性了。

核心要点就三个：

• 认证与鉴权：API Key + Secret 是标配，有些还要OAuth2.0。我习惯把密钥放在环境变量里，绝不硬编码。
• 限流与重试：用令牌桶或滑动窗口控制请求频率。遇到429（Too Many Requests）要指数退避重试。
• 数据格式解析：JSON/XML/Protobuf，金融行情常用Protobuf，解析效率高很多。

import aiohttp
import asyncio

async def fetch_price(session, symbol):
    url = f"https://api.example.com/v1/quote/{symbol}"
    async with session.get(url) as resp:
        data = await resp.json()
        return data['price']

async def main():
    symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT']
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_price(session, s) for s in symbols]
        prices = await asyncio.gather(*tasks)
        print(prices)

asyncio.run(main())

数据库连接：MySQL vs PostgreSQL 怎么选？

金融数据存储，MySQL和PostgreSQL是两大主流。我个人习惯：高频交易日志用MySQL，复杂分析查询用PostgreSQL。为什么？

对比维度	MySQL	PostgreSQL
写入性能	高（InnoDB引擎优化好）	中等（但支持并行写入）
复杂查询	一般	强（窗口函数、CTE、数组类型）
金融时序支持	需额外设计分区表	原生支持TimescaleDB扩展
连接池	推荐PyMySQL + DBUtils	推荐psycopg2 + SQLAlchemy

连接数据库时，我建议用连接池而不是每次新建连接。你想想看，金融数据ETL任务动不动就跑几百万条，每次新建连接的开销太浪费了。

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# MySQL连接池
mysql_engine = create_engine(
    'mysql+pymysql://user:pass@host/db',
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)

# PostgreSQL连接池
pg_engine = create_engine(
    'postgresql+psycopg2://user:pass@host/db',
    pool_size=10,
    max_overflow=20
)

CSV/Excel批量导入：看似简单，坑最多

CSV和Excel是金融数据交换的「通用语言」。但说实话，我处理过的CSV文件里，十有八九都有问题——编码不对、分隔符不统一、日期格式乱七八糟。

我的标准处理流程：

1. 检测编码：用chardet库自动检测，避免乱码
2. 统一分隔符：CSV不一定是逗号，也可能是制表符或分号
3. 类型推断：日期、数值、字符串要明确指定，别让pandas猜
4. 分批写入：大文件不要一次性读入内存，用chunksize分批处理

import pandas as pd
from sqlalchemy import text

# 分批读取CSV并写入数据库
chunksize = 10000
for chunk in pd.read_csv('trade_data.csv', chunksize=chunksize, encoding='utf-8'):
    # 类型转换
    chunk['trade_date'] = pd.to_datetime(chunk['trade_date'])
    chunk['amount'] = chunk['amount'].astype(float)
    # 写入数据库
    chunk.to_sql('trades', engine, if_exists='append', index=False)
    print(f"已写入 {len(chunk)} 条")

增量更新策略：别每次都全量跑

金融数据是持续产生的，每天全量更新太傻了。我常用的增量策略有三种：

• 时间戳增量：记录上次同步时间，只拉取该时间之后的数据。最简单，但要求源数据有时间字段。
• 偏移量增量：用自增ID或序列号，记录上次处理到的位置。适合日志类数据。
• 变更数据捕获（CDC）：监听数据库的binlog或WAL日志，实时捕获变更。适合对实时性要求高的场景。

我个人最常用的是时间戳增量，配合一个「增量状态表」来记录每个数据源的同步进度。

-- 增量状态表
CREATE TABLE sync_status (
    source_name VARCHAR(100) PRIMARY KEY,
    last_sync_time TIMESTAMP,
    last_offset BIGINT,
    status VARCHAR(20)
);

-- 查询上次同步时间
SELECT last_sync_time FROM sync_status WHERE source_name = 'trade_api';

增量更新还有个关键点——去重。金融数据经常有重复推送，我习惯在写入前用主键或唯一索引做upsert（插入或更新）。PostgreSQL的ON CONFLICT语法很好用，MySQL也有ON DUPLICATE KEY UPDATE。

-- PostgreSQL upsert
INSERT INTO trades (trade_id, symbol, price, volume, trade_time)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (trade_id) 
DO UPDATE SET price = EXCLUDED.price, volume = EXCLUDED.volume;

知识体系总览

下面这张图，是我梳理的数据源接入与ETL核心逻辑。你可以把它当作一个检查清单——每次接入新数据源时，对着走一遍，基本不会漏。

这张图把整个流程分成了四层：数据源层、采集连接层、转换清洗层、存储层。增量策略贯穿其中，不是单独一个环节。你想想看，如果每次采集都全量拉，那连接层和转换层的压力会非常大。所以增量策略要从源头就开始设计。