## 开篇：每个美股数据源都有自己的“脾气”

做美股量化的朋友，一定遇到过这些场景：

- 用 **yfinance** 拉数据，发现 `Adj Close` 和 `Close` 有时候一样，有时候不一样，到底该信哪个？
- 用 **akshare** 拿美股历史，返回的列名全是中文（`开盘`、`收盘`），而且分页还只能拿 200 条。
- 用 **Polygon.io** 的聚合接口，发现 1 分钟 K 线居然“跳分钟”了，是不是数据丢了？

每个数据源都有自己的“脾气”。如果你没有一套统一的清洗框架，每次换源都要重写一遍解析逻辑，不仅浪费时间，还容易埋下数据质量隐患。

另外，有读者问到美股数据源的安全性。与国内部分依赖爬虫、接口不稳定的数据源不同，主流美股数据源（如 Polygon、TickDB）普遍提供标准的 REST/WebSocket API，采用 TLS 加密传输和 API Key 认证，数据传输安全性更高。受 SEC 监管的数据服务商在数据完整性、服务可用性（SLA）方面也有更严格的承诺。

今天这篇文章，我们就以 **yfinance、akshare、Polygon.io** 三个主流美股数据源为例，手把手教你如何构建一套**通用的数据清洗管道**。内容包括：
- 字段映射：把千奇百怪的字段名统一成标准模型
- 时间对齐：处理时区、时间戳格式，以及不同频率的对齐技巧
- 复权处理：拆股、分红后价格怎么调
- 缺失值处理：分钟数据缺失、节假日跳空
- 统一清洗框架：适配器模式 + 交易日历 + 健壮的错误处理

最后，我们会聊聊如何通过 TickDB 这类内置标准化能力的数据源，省去 90% 的清洗代码。

---

## 一、先定义你的“标准模型”

无论接入哪个数据源，最终都要转换成内部统一的格式。推荐以下标准模型（以 K 线为例）：

| 字段名 | 类型 | 说明 | 单位/格式 |
|--------|------|------|----------|
| `symbol` | string | 统一 ticker（如 `AAPL.US`） | 带市场后缀 |
| `timestamp` | int | Unix 毫秒时间戳 | UTC |
| `open` | float | 开盘价 | 美元/股 |
| `high` | float | 最高价 | 美元/股 |
| `low` | float | 最低价 | 美元/股 |
| `close` | float | 收盘价 | 美元/股 |
| `volume` | int | 成交量 | 股 |

有了这个标准，后续每个数据源的适配器只需要做一件事：**把源数据映射到这个模型**。

---

## 二、三个美股数据源的真实“乱象”与清洗方案

### 2.1 Yahoo Finance（yfinance）

#### 原始数据结构
yfinance 是目前最流行的免费美股数据 Python 库。它返回的是一个 Pandas DataFrame，典型字段如下（以日线为例）：

| Date       | Open    | High    | Low     | Close   | Adj Close | Volume   |
|------------|---------|---------|---------|---------|-----------|----------|
| 2025-04-01 | 175.00  | 176.50  | 174.80  | 176.20  | 176.20    | 12345678 |

- **字段名**：首字母大写，`Adj Close` 带空格。
- **时间索引**：Datetime 对象，默认时区为美东时间（EST/EDT）。
- **成交量**：单位是“股”。
- **复权**：提供 `Adj Close` 列，但 `Open`、`High`、`Low`、`Close` 默认是**已复权**的（`auto_adjust=True`），`Adj Close` 实际上是冗余列。

#### 常见问题与清洗要点

**① 复权字段混淆**  
当设置 `auto_adjust=False` 时，会返回原始未复权价格，以及单独的 `Adj Close` 列。很多用户发现 `Adj Close` 和 `Close` 在某些时间段完全相同，以为出了 bug。实际上，这是因为该时间段内没有发生拆股或分红，复权价自然等于未复权价。**清洗时需明确：默认使用 `Close`（已复权），无需额外处理。**

**② 时区问题**  
yfinance 返回的时间索引是带时区的（美东时间）。如果你在 UTC 时区的服务器上运行，直接与其他数据源合并会导致错位。解决方案：统一转换为 UTC 毫秒时间戳。

```python
# 将 yfinance DataFrame 的索引转为 UTC 毫秒
df.index = df.index.tz_convert('UTC')
df['timestamp'] = df.index.astype('int64') // 10**6
```

**③ 缺失数据**  
部分历史日期（尤其是早于 2000 年的数据）可能缺失。yfinance 不会自动填充，需在清洗层处理。对于回测场景，建议使用前向填充（`ffill`）或直接剔除。

**清洗代码示例（yfinance → 标准模型）**：

```python
import yfinance as yf
import pandas as pd

def clean_yfinance(ticker, start, end):
    raw = yf.download(ticker, start=start, end=end, auto_adjust=True)
    # 重置索引，将日期转为列
    df = raw.reset_index()
    # 时间戳转换：美东时间 -> UTC 毫秒
    df['timestamp'] = pd.to_datetime(df['Date']).dt.tz_localize('US/Eastern').dt.tz_convert('UTC').astype('int64') // 10**6
    # 重命名并选择标准字段
    df = df.rename(columns={
        'Open': 'open',
        'High': 'high',
        'Low': 'low',
        'Close': 'close',
        'Volume': 'volume'
    })
    df['symbol'] = ticker
    return df[['symbol', 'timestamp', 'open', 'high', 'low', 'close', 'volume']]
```

---

### 2.2 akshare（美股部分）

#### 原始数据结构
akshare 是一个国内开源的金融数据库，通过聚合多个公开数据源（如东方财富、新浪）提供接口。以美股历史数据为例，使用 `stock_us_hist` 接口（东方财富源），返回的 DataFrame 字段为中文：

| 日期       | 开盘   | 收盘   | 最高   | 最低   | 成交量   | 成交额   | ... |
|------------|--------|--------|--------|--------|----------|----------|-----|
| 2025-04-01 | 175.00 | 176.20 | 176.50 | 174.80 | 12345678 | 2.17e9   | ... |

- **字段名**：全中文。
- **时间**：字符串格式 `YYYY-MM-DD`。
- **成交量**：经实测，`stock_us_hist` 返回的成交量单位为“股”，但其他接口可能为“手”，建议在清洗层通过配置化方式处理。
- **复权**：akshare 的 `stock_us_hist` 接口支持 `adjust` 参数，可设为 `'qfq'`（前复权）或 `'hfq'`（后复权）。默认不包含复权，需要显式指定。

#### 常见问题与清洗要点

**① 中文列名**  
直接使用中文列名在代码中不优雅，且容易出错。必须映射为英文标准字段。

**② 分页限制**  
部分接口（如获取美股列表）默认每页只能返回 200 条，需要循环分页才能获取完整数据。对于历史 K 线，一般不存在分页问题。

**③ 复权支持**  
akshare 的美股接口支持复权参数，但不同底层源（东方财富、新浪）的返回格式可能略有差异，建议使用时查阅最新文档。

**清洗代码示例（akshare → 标准模型）**：

```python
import akshare as ak
import pandas as pd

def clean_akshare(symbol):
    # 示例：获取美股日线，symbol 如 'AAPL'，使用前复权
    raw = ak.stock_us_hist(symbol=symbol, period='daily', adjust='qfq')
    # 中文列名映射
    raw = raw.rename(columns={
        '日期': 'date',
        '开盘': 'open',
        '收盘': 'close',
        '最高': 'high',
        '最低': 'low',
        '成交量': 'volume'
    })
    # 时间戳转换：日期字符串 -> UTC 毫秒（假设数据为美东时间收盘）
    raw['timestamp'] = pd.to_datetime(raw['date']).dt.tz_localize('US/Eastern').dt.tz_convert('UTC').astype('int64') // 10**6
    raw['symbol'] = symbol
    return raw[['symbol', 'timestamp', 'open', 'high', 'low', 'close', 'volume']]
```

**注意**：akshare 的美股数据准确性依赖底层源（东方财富），建议与 yfinance 交叉验证。

---

### 2.3 Polygon.io

#### 原始数据结构
Polygon.io 是一个专业的金融数据 API 服务商，提供 REST 和 WebSocket 接口。其聚合（Aggregates）端点返回的 JSON 格式如下：

```json
{
  "ticker": "AAPL",
  "results": [
    {"t": 1711910400000, "o": 175.00, "h": 176.50, "l": 174.80, "c": 176.20, "v": 12345678}
  ]
}
```

- **字段**：`t`（时间戳，毫秒）、`o`、`h`、`l`、`c`、`v`（成交量）。
- **时间戳**：Unix 毫秒，UTC。
- **成交量**：单位是“股”。
- **复权**：可通过参数 `adjusted=true/false` 控制是否返回复权数据。

#### 常见问题与清洗要点

**① 分钟数据“缺失”**  
很多用户发现 Polygon 的 1 分钟聚合数据不连续，比如 10:01 有数据，10:02 没有，10:03 又有。实际上，这是因为在 10:02 这一分钟内**没有任何交易发生**（Volume=0），Polygon 不会生成该分钟的聚合记录。这不是错误，是设计如此。清洗时需要用前向填充（forward-fill）补全缺失的分钟。

**② 复权计算潜在误差**  
有部分用户在 Reddit 等社区反馈，在极端拆股场景下（如 1:10 反向拆股），Polygon 的 `adjusted=true` 返回的价格可能存在计算误差。建议在生产环境中使用 `adjusted=false` 获取原始价格，并结合独立的拆股数据库自行计算复权，或通过交叉验证确认数据准确性。

**③ 实时数据乱序**  
WebSocket 推送的实时 tick 可能因网络延迟而乱序到达。不能依赖本地接收时间排序，必须使用消息内的 `sip_timestamp`（交易所撮合时间）进行排序。

**清洗代码示例（Polygon REST → 标准模型）**：

```python
import requests
import pandas as pd

def clean_polygon(ticker, api_key):
    url = f"https://api.polygon.io/v2/aggs/ticker/{ticker}/range/1/day/2025-01-01/2025-04-01?adjusted=false&apiKey={api_key}"
    resp = requests.get(url).json()
    records = []
    for item in resp['results']:
        records.append({
            'symbol': ticker,
            'timestamp': item['t'],  # 已经是毫秒
            'open': item['o'],
            'high': item['h'],
            'low': item['l'],
            'close': item['c'],
            'volume': item['v']
        })
    df = pd.DataFrame(records)
    return df
```

**前向填充缺失分钟**（针对分钟数据）：

```python
# 假设 df 是分钟数据，时间戳列 'timestamp'（毫秒）
df.set_index('timestamp', inplace=True)
full_index = pd.date_range(start=df.index.min(), end=df.index.max(), freq='1min')
df = df.reindex(full_index).fillna(method='ffill')
df['volume'] = df['volume'].fillna(0)
```

---

## 三、时间对齐的进阶处理

对于日线数据，对齐到交易日历即可；对于分钟数据，需要注意交易所的休市时间段（如午休、提前收盘）。可以使用 `pandas_market_calendars` 获取交易所的 `schedule`，其中包含 `market_open` 和 `market_close` 时间，据此过滤非交易时段的数据。

```python
import pandas_market_calendars as mcal

nyse = mcal.get_calendar('NYSE')
schedule = nyse.schedule(start_date='2025-01-01', end_date='2025-12-31')
# 生成标准的交易日历（美东时间日期）
trading_days = nyse.valid_days(schedule)

# 将日期转为 UTC 毫秒时间戳（当日 00:00:00 UTC）
trading_timestamps = [int(pd.Timestamp(d).tz_localize('US/Eastern').tz_convert('UTC').timestamp() * 1000) for d in trading_days]

# 将数据 reindex 到交易日历
df = df.set_index('timestamp').reindex(trading_timestamps).fillna(method='ffill')
```

---

## 💡 架构师笔记：WebSocket 实时流的时间戳对齐

在构建实时数据管道时，最容易被忽视的是 WebSocket 流的数据对齐。不同数据源的时间戳含义不同：
- **Polygon.io**：使用 `sip_timestamp`（交易所撮合时间，纳秒精度）
- **某些免费源**：使用本地接收时间（受网络延迟影响）

**切记**：必须按照交易所时间戳排序，而不是本地接收时间。否则在网络抖动时，数据会乱序到达，导致 K 线计算错误。

---

## 四、统一清洗框架：适配器模式 + 交易日历 + 健壮错误处理

### 4.1 适配器模式（Adapter Pattern）

为了优雅地支持多个数据源，建议采用**适配器模式**：定义一个统一的数据获取接口，每个数据源实现自己的适配器，内部完成字段映射、时间转换、复权处理。

```python
from abc import ABC, abstractmethod
import pandas as pd

class DataSourceAdapter(ABC):
    @abstractmethod
    def get_ohlcv(self, symbol, start, end, interval='1d') -> pd.DataFrame:
        pass

class YFinanceAdapter(DataSourceAdapter):
    def get_ohlcv(self, symbol, start, end, interval='1d'):
        # 调用 yfinance，清洗后返回标准 DataFrame
        pass

class AkshareAdapter(DataSourceAdapter):
    def get_ohlcv(self, symbol, start, end, interval='1d'):
        # 调用 akshare，清洗后返回标准 DataFrame
        pass

class PolygonAdapter(DataSourceAdapter):
    def get_ohlcv(self, symbol, start, end, interval='1d'):
        # 调用 Polygon API，清洗后返回标准 DataFrame
        pass
```

这样，上层业务代码只需调用统一的 `get_ohlcv`，切换数据源时只需更换适配器，无需修改任何清洗逻辑。

### 4.2 健壮的错误处理与重试机制

在实际生产环境中，网络超时、API 限流、空数据等情况非常常见。下面是一个带重试和限流处理的请求示例（使用 `tenacity` 库）：

```python
import time
import requests
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_with_retry(url, headers=None, timeout=10):
    resp = requests.get(url, headers=headers, timeout=timeout)
    if resp.status_code == 429:
        retry_after = int(resp.headers.get('Retry-After', 5))
        time.sleep(retry_after)
        raise Exception("Rate limited")
    resp.raise_for_status()
    return resp.json()
```

---

## 五、不想自己清洗？试试内置标准化的数据源

数据清洗虽然必要，但非常繁琐。如果你希望开箱即用，可以考虑像 **TickDB** 这类内置了标准化的数据源。TickDB 的特点包括：

- **统一接口**：REST 和 WebSocket 返回相同的数据结构，字段名、单位、时间戳格式完全统一。
- **自动复权**：K 线接口默认返回前复权数据，无需手动计算。
- **多市场覆盖**：同时支持美股、港股、A股、加密货币，且 ticker 命名规范（如 `AAPL.US`、`700.HK`、`000001.SZ`）。
- **AI 友好**：内置 Skill 能力，可无缝集成到 AI 应用中。
- **生产级 WebSocket**：如果接入 TickDB 的 WebSocket 实时流（`wss://api.tickdb.ai/v1/realtime`），记得在客户端实现每秒发送 `{"cmd": "ping"}` 的心跳保活，这是保障高可用的基本操作。

对于生产级系统，使用 TickDB 可以节省 90% 的清洗代码，让你专注于策略本身。当然，如果你喜欢自己掌控细节，ClawHub 上也有不少开源清洗工具可以参考。

---

## 六、总结

本文以 yfinance、akshare、Polygon.io 三个主流美股数据源为例，详细讲解了：
- 各数据源的**原始数据结构**与常见“坑”
- **字段映射、时间对齐（含交易日历）、复权处理、缺失值填充**的具体代码
- 如何用**适配器模式**构建统一清洗框架
- 健壮的**错误处理与重试机制**
- 美股数据源的**安全性**优势

无论你选择哪个数据源，一套健壮的清洗管道都是量化系统的地基。希望这篇文章能帮你少走弯路，把时间花在策略上，而不是数据格式上。

> 本文纯技术分享，提到的数据源均为公开服务，不构成任何投资建议。市场有风险，投资需谨慎。

用 yfinance 拉数据，发现 Adj Close 和 Close 有时候一样，有时候不一样，到底该信哪个？

Yahoo Finance、akshare、Polygon 数据太乱怎么办？美股行情清洗实战：字段映射、时间对齐、复权处理

如何构建一套**通用的数据清洗管道**

后端

大数据

职业发展

美股量化交易中，yfinance、akshare和Polygon.io三大数据源各有特点，但数据格式差异大。本文详解如何构建统一清洗框架，包括字段映射、时间对齐、复权处理等核心技巧，帮助量化开发者高效处理数据，提升策略开发效率。

数据完整性

加密货币

数据清洗

网络延迟

数据管道

数据结构

服务器

数据库

Python

JSON

Unix

2026采购季 | AI焕新·智启新局

tencentdb-catalog

4核4G3M云服务器 新用户低至38元/年！

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Yahoo Finance、akshare、Polygon 数据太乱怎么办？美股行情清洗实战：字段映射、时间对齐、复权处理-腾讯云开发者社区-腾讯云

Yahoo Finance、akshare、Polygon 数据太乱怎么办？美股行情清洗实战：字段映射、时间对齐、复权处理

Yahoo Finance、akshare、Polygon 数据太乱怎么办？美股行情清洗实战：字段映射、时间对齐、复权处理

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐