etf/framework_v2/FLASK_API_FETCHER_GUIDE.md

FlaskAPIFetcher 使用指南

概述

FlaskAPIFetcher 是 framework_v2 的数据获取器实现，通过 HTTP API 获取线上数据（指数、ETF）。

核心优势：

• ✅ 无需本地 SSH 隧道配置
• ✅ 支持远程调用（生产环境）
• ✅ 自动重试 + 超时处理
• ✅ Pydantic Schema 验证响应
• ✅ ETF 数据自动附加净值和溢价率

---

快速开始

1. 基础使用

from framework_v2.shared.data import FlaskAPIFetcher

# 创建数据获取器
fetcher = FlaskAPIFetcher(
    base_url="https://k3s.tokenpluse.xyz",  # 或从环境变量读取
    timeout=120,
    retries=3
)

# 获取指数数据
data = fetcher.fetch_indices(
    codes=["000300.SH", "000905.SH"],
    start="2024-01-01",
    end="2024-12-31"
)

# 访问数据
df_300 = data["000300.SH"]
print(df_300.head())

2. 获取 ETF 数据

# 获取 ETF 数据（自动附加净值和溢价率）
data = fetcher.fetch_etf(
    codes=["510300.SH", "159919.SZ"],
    start="2024-01-01",
    end="2024-12-31"
)

# 访问价格数据
df = data["510300.SH"]
print(df.head())

# 访问净值数据
nav = df.attrs.get('nav')
if nav is not None:
    print(f"净值数据: {len(nav)} 条")

# 访问溢价率
premium = df.attrs.get('latest_premium')
if premium is not None:
    print(f"最新溢价率: {premium:.2f}%")

---

完整示例：结合 CrossMarketAligner

场景：获取跨市场数据并对齐到 A 股日历

from framework_v2.shared.data import FlaskAPIFetcher, CrossMarketAligner

# 1. 创建数据获取器
fetcher = FlaskAPIFetcher()

# 2. 获取 A 股交易日历
a_share_calendar = fetcher.get_trading_calendar(market='A')

# 3. 创建对齐器
aligner = CrossMarketAligner(target_calendar=a_share_calendar)

# 4. 获取跨市场指数数据
us_indices = fetcher.fetch_indices(
    codes=["^GSPC", "^IXIC"],  # 美股
    start="2024-01-01",
    end="2024-12-31"
)

cn_indices = fetcher.fetch_indices(
    codes=["000300.SH", "000905.SH"],  # A股
    start="2024-01-01",
    end="2024-12-31"
)

# 5. 对齐收益率到 A 股日历
returns_aligned = aligner.align_multi_asset(
    close_dict={
        "SP500": us_indices["^GSPC"]["close"],
        "NASDAQ": us_indices["^IXIC"]["close"],
        "CSI300": cn_indices["000300.SH"]["close"],
        "CSI500": cn_indices["000905.SH"]["close"],
    }
)

# 6. 验证对齐结果
print(returns_aligned.head())
print(f"\nNaN 数量: {returns_aligned.isna().sum().sum()}")  # 应该为 0

---

API 参考

FlaskAPIFetcher

初始化

FlaskAPIFetcher(
    base_url: str = None,      # API 地址（默认从环境变量读取）
    timeout: int = 120,        # 请求超时时间（秒）
    retries: int = 3           # 重试次数
)

核心方法

方法	说明	返回类型
fetch_indices(codes, start, end)	获取指数 OHLCV 数据	Dict[str, DataFrame]
fetch_etf(codes, start, end)	获取 ETF 数据（价格+净值）	Dict[str, DataFrame]
get_trading_calendar(market)	获取交易日历	pd.Index
get_benchmark(code, start, end)	获取基准数据	pd.Series
get_health()	检查 API 健康状态	Dict

fetch_indices 参数

fetcher.fetch_indices(
    codes=["000300.SH", "000905.SH"],  # 指数代码列表
    start="2024-01-01",                # 开始日期
    end="2024-12-31"                   # 结束日期
)

返回 DataFrame 结构：

            code     open     high      low    close     volume
date
2024-01-02  000300.SH  3388.30  3395.40  3372.50  3390.20  12345678
2024-01-03  000300.SH  3390.20  3405.60  3385.10  3398.50  13456789

fetch_etf 参数

fetcher.fetch_etf(
    codes=["510300.SH", "159919.SZ"],  # ETF 代码列表
    start="2024-01-01",                # 开始日期
    end="2024-12-31"                   # 结束日期
)

返回 DataFrame 结构：

            code     open     high      low    close     volume
date
2024-01-02  510300.SH  3.520    3.545    3.510    3.540    45678901

附加信息（df.attrs）：

• nav: 净值数据 DataFrame
• premium_series: 溢价率序列（dict）
• latest_premium: 最新溢价率（float）
• premium_stats: 溢价率统计（dict）

---

与 DataFetcher 抽象基类的关系

framework_v2/core/data.py              # 抽象基类
    └── DataFetcher (ABC)
        ├── fetch_indices() [抽象]
        ├── fetch_etf() [抽象]
        ├── get_trading_calendar() [抽象]
        └── get_benchmark() [可选]

framework_v2/shared/data/flask_api_fetcher.py  # 具体实现
    └── FlaskAPIFetcher(DataFetcher)
        ├── fetch_indices() ✅ 实现（调用 FlaskAPIDataSource）
        ├── fetch_etf() ✅ 实现（调用 FlaskAPIDataSource）
        ├── get_trading_calendar() ✅ 实现（临时：pandas BDay）
        └── get_benchmark() ✅ 实现

继承关系验证

from framework_v2.core.data import DataFetcher
from framework_v2.shared.data import FlaskAPIFetcher

# 验证继承
assert issubclass(FlaskAPIFetcher, DataFetcher)

# 验证抽象方法已实现
fetcher = FlaskAPIFetcher()
assert hasattr(fetcher, 'fetch_indices')
assert hasattr(fetcher, 'fetch_etf')
assert hasattr(fetcher, 'get_trading_calendar')

---

环境变量配置

FLASK_API_URL

# .env 文件
FLASK_API_URL=https://k3s.tokenpluse.xyz

优先级：

1. 构造函数参数 base_url
2. 环境变量 FLASK_API_URL
3. 默认值 https://k3s.tokenpluse.xyz

---

错误处理

自动重试

fetcher = FlaskAPIFetcher(retries=3)

# 失败时自动重试：
# - 网络超时
# - HTTP 5xx 错误
# - JSON 解析失败

手动错误处理

data = fetcher.fetch_indices(["000300.SH"], "2024-01-01", "2024-12-31")

if "000300.SH" not in data:
    print("✗ 数据获取失败")
    # 处理错误...
else:
    print(f"✓ 获取 {len(data['000300.SH'])} 条数据")

---

性能优化

批量获取 vs 单个获取

# ✅ 推荐：批量获取（内部自动重试 + 进度显示）
data = fetcher.fetch_indices(
    codes=["000300.SH", "000905.SH", "000852.SH"],
    start="2024-01-01",
    end="2024-12-31"
)

# ❌ 不推荐：循环单个获取（无进度显示）
for code in codes:
    df = fetcher._source.fetch(code, start, end)

超时设置

# 网络较慢时增加超时
fetcher = FlaskAPIFetcher(timeout=180)  # 3 分钟

---

测试

运行测试验证功能：

cd /Users/aszer/Documents/vscode/etf
python framework_v2/tests/test_flask_api_fetcher.py

预期输出：

✓ 测试通过 - 健康检查
✓ 测试通过 - 指数数据
✓ 测试通过 - ETF 数据
✓ 测试通过 - 交易日历
✓ 测试通过 - 基准数据

总计: 5/5 通过

---

相关文档

• 框架总览 - framework_v2 架构说明
• 数据架构方案 - 数据流设计
• 跨市场对齐方案 - CrossMarketAligner 使用
• Aligner + Schema 整合 - 验证架构

---

注意事项

1. 交易日历准确性

当前 get_trading_calendar() 使用 pandas bdate_range 生成近似日历，未考虑节假日。

临时方案：

calendar = fetcher.get_trading_calendar(market='A')
# 手动移除节假日
holidays = pd.to_datetime(['2024-02-10', '2024-10-01', ...])
calendar = calendar[~calendar.isin(holidays)]

TODO：后续通过 API 端点获取准确日历。

2. ETF 净值数据量

ETF 净值数据可能远多于价格数据（历史净值 vs 交易价格）：

df = data["510300.SH"]
print(f"价格: {len(df)} 条")        # ~60 条（2024 Q1）
print(f"净值: {len(df.attrs['nav'])} 条")  # ~3695 条（全历史）

3. 资产类型检测

FlaskAPIDataSource 支持自动检测资产类型，也可手动指定：

# 自动检测
df = fetcher._source.fetch("510300.SH", start, end)

# 手动覆盖
df = fetcher._source.fetch("510300.SH", start, end, asset_type='china_etf')

---

版本历史

• 2024-04-16: 初始版本
  • 继承 DataFetcher 抽象基类
  • 实现指数、ETF 数据获取
  • 集成 FlaskAPIDataSource
  • 5/5 测试通过