阅读提示

这篇是 Datafeed 模块的实战手册，讲解交易日生成、可交易池过滤、行情/因子查询的常用 API 与语义边界。学完这篇，你能在任何需要自定义数据查询的场景里快速写出正确的代码。

导言

Datafeed 是 Betalens 通往 PostgreSQL 的唯一大门。所有数据——无论是股价、股息率、换手率、行业分类——都要通过 Datafeed 的 API 查询。理解这些 API，能让你在框架提供的高层函数（pre_query_characteristic_data、single_characteristic）之外，也能在需要时直接写自定义查询。

配置与连接

Datafeed 的数据库连接遵循以下优先级（高 → 低）：

  1. 运行时传入的 db_config
  2. BETALENS_DB_* 环境变量
  3. BETALENS_CONFIG 指定的配置文件
  4. ~/.config/betalens/config.json
  5. betalens/datafeed/config.local.json

通常只需要配置 config.local.json（见 03 篇），不需要改其他地方：

1
2
3
4
from betalens.datafeed.config import get_config

config = get_config()
print(config.get("database.dbname"))  # 通常是 "datafeed"

交易日生成：get_absolute_trade_days

这是使用频率最高的函数之一，用于生成调仓日序列。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from betalens.datafeed import get_absolute_trade_days, trade_days_offset

# === period 参数：生成周期 ===
days_Y = get_absolute_trade_days("2020-01-01", "2024-12-31", "Y")
# 返回：每年最后或最近的交易日（对齐年报披露截止日 4月30日）
print(f"年度调仓日：{len(days_Y)} 个")
# [Timestamp('2020-04-29'), Timestamp('2021-04-29'), ...]

days_M = get_absolute_trade_days("2024-01-01", "2024-12-31", "M")
# 返回：每月最后交易日
print(f"月度调仓日：{len(days_M)} 个")  # ~12 个

days_W = get_absolute_trade_days("2024-01-01", "2024-12-31", "W")
# 返回：每周最后交易日
print(f"周度调仓日：{len(days_W)} 个")  # ~52 个

days_D = get_absolute_trade_days("2024-01-01", "2024-12-31", "D")
# 返回：每日交易日（排除周末和节假日）
print(f"日度调仓日：{len(days_D)} 个")  # ~244 个

# === exchange 参数：交易所 ===
days_SH = get_absolute_trade_days("2024-01-01", "2024-12-31", "D", exchange="SHSE")
# 默认是 "SHSE"（上海交易所），会自动覆盖沪深
days_NIB = get_absolute_trade_days("2024-01-01", "2024-12-31", "D", exchange="NIB")
# NIB = 新三板（北交所）

# === 日期偏移 ===
next_day = trade_days_offset("2024-01-31", 1, period="D")
# "2024-01-31" 后第一个交易日
prev_week = trade_days_offset("2024-01-15", -1, period="W")
# "2024-01-15" 前一个周末（周五）

period 支持的取值：

值 含义 典型用法
D 每日交易日 日频策略
W 每周（最后交易日） 周频调仓
M 每月（最后交易日） 月频调仓
Q 每季度末 季频因子（如财务数据）
S 每半年末 半年调仓
Y 每年（4月30日或最近交易日） 对齐年报披露

注意：get_absolute_trade_days 只从本地的 trade_calendar_day 表读取，不访问任何在线服务。第一次使用前需要通过 betalens_db_manager 导入交易日历。

可交易池：get_tradable_pool

这个函数基于 trade_status（交易状态）过滤，返回哪些股票在哪些日期是”可以正常买卖的”。

1
2
3
4
5
6
7
from betalens.factor.factor import get_tradable_pool

# days 是 get_absolute_trade_days 的返回值
date_ranges, code_ranges = get_tradable_pool(days)
# date_ranges: DataFrame，shape = (len(days), n_codes)
# code_ranges: DataFrame，shape = (n_codes,)
# True = 可交易，False = 停牌/退市/未上市

get_tradable_pool 返回两个对象：

  • date_ranges：日期 × 股票代码矩阵，标记每天哪些股票可交易。
  • code_ranges：哪些股票在整个回测区间内有任何可交易日。

这两个对象会传给 pre_query_characteristic_data 的 date_ranges 和 code_ranges 参数，确保查询时只取可交易证券的数据。

默认行为（include_abnormal=False）：只保留”正常交易”的证券，即 trade_status = 1 的证券。停牌（0）和未上市（-1）的证券被排除。

include_abnormal=True：包含退市/停牌证券，主要用于事后分析或特殊场景（如想看停牌期间的因子值变化）。日常因子回测不要开这个选项。

1
2
# 用于事后分析场景（看停牌证券的历史因子值）
date_ranges_ab, code_ranges_ab = get_tradable_pool(days, include_abnormal=True)

Datafeed 查询：行情

1
2
3
4
5
6
7
8
9
10
11
12
13
from betalens.datafeed import Datafeed

data = Datafeed("daily_market")  # 对应 market_daily_fact

# 查询时间范围
prices = data.query_time_range(
    codes=["000001.SZ", "000002.SZ"],
    start_date="2024-01-01",
    end_date="2024-01-31",
    metric="收盘价(元)",
)
print(prices.head())
data.close()

常用表名映射：

表名 内容 常见 metric
daily_market A股/港股日行情 收盘价(元)、成交量(股)、成交额(元)
daily_index 指数日行情 收盘价(元)（用于 benchmark）
daily_fund 基金日行情 收盘价(元)、净值(元)
daily_bond 债券日行情 收盘价(元)
fundamentals 财务因子 股息率(报告期)、ROE(报告期)、市净率
industry 行业分类 行业代码、行业名称

PIT 查询：query_nearest_before / query_nearest_after

这是做因子研究时最重要的查询方式——在某观测时点，找到最近一个可用的数据点。

1
2
3
4
5
6
7
8
9
10
data = Datafeed("fundamentals")

df = data.query_nearest_before({
    "codes": ["000001.SZ", "000002.SZ"],
    "datetimes": ["2024-04-30 15:00:01", "2024-08-31 15:00:01"],
    "metric": "股息率(报告期)",
    "time_tolerance": 24 * 365,  # 最多往前找1年内的数据
})
print(df)
data.close()

返回结果大概是：

code query_date trade_date value
000001.SZ 2024-04-30 2024-04-29 3.24
000001.SZ 2024-08-31 2024-04-29 3.24
000002.SZ 2024-04-30 2024-04-29 2.81
000002.SZ 2024-08-31 2024-08-30 2.95
  • query_date 是你做决策的日期。
  • trade_date 是实际找到的数据的披露日（满足 trade_date <= query_date 且 value_end_date <= query_date）。
  • time_tolerance 如果设小了（比如 24*30），超过 30 天没找到数据就返回空。

query_nearest_after：找某日期之后最近的数据，常用于分红送转等”未来已知”事件的处理（不常用，了解即可）。

行业查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from betalens.datafeed import query_industry, get_industry_members

# 查询某日期某股票的行业
ind = query_industry(
    codes=["000001.SZ", "000002.SZ"],
    dates=["2024-04-30", "2024-12-31"],
    scheme="申万一级行业",
)
print(ind)

# 查询某行业在某日期的所有成分股
members = get_industry_members(
    scheme="申万一级行业",
    industry_name="银行",
    dates=["2024-04-30"],
)
print(f"银行行业成分股数量：{len(members)}")

开发者侧：为什么要用 Datafeed 而不直接写 SQL

有几个实际原因：

  1. 统一连接管理：不需要每次查询都写连接池初始化代码。
  2. PIT 语义封装：不用自己写 value_end_date 比较逻辑。
  3. metric 映射：用中文 metric 名（如 "收盘价(元)"）查询，Datafeed 内部自动做 metric_alias 映射。
  4. 懒加载/批量优化：内部有批量查询优化，减少数据库往返次数。

如果你确实需要直接写 SQL，Datafeed 对象有一个 .conn 属性可以访问原始连接，但不推荐在研究脚本里直接写 SQL——除非你在开发 Datafeed 本身的功能。

常见错误

1. 交易日历为空

1
2
days = get_absolute_trade_days("2020-01-01", "2024-12-31", "Y")
# 返回空列表 []

原因：本地 trade_calendar_day 表没有数据。用 betalens_db_manager 导入日历数据：

1
python -m betalens_db_manager import-calendar

2. metric 名写错

1
2
3
4
5
# 报错：metric not found
data.query_time_range(..., metric="收盘价")

# 对：中文名必须与数据库一致
data.query_time_range(..., metric="收盘价(元)")

可以用 Datafeed 的 get_metric_names() 列出所有可用 metric。

3. 忘记 data.close()

Datafeed 使用连接池，查询完记得关闭，避免连接泄漏：

1
2
3
4
5
6
7
8
data = Datafeed("daily_market")
try:
    result = data.query_time_range(...)
finally:
    data.close()
# 或用上下文管理器：
with Datafeed("daily_market") as data:
    result = data.query_time_range(...)

延伸阅读