版本 0.2.1最低支持核心版本： >=0.4.1

DataManager

描述： 统一数据访问门面（进程内默认单例）

init / initialize

• 类型：构造 / 实例方法
• 状态：beta
• 描述： 构造时幂等 initialize()（锁保护）；默认单例，force_new=True 可破例
• 参数（构造）：
  • db：可选已有 DatabaseManager
  • is_verbose：日志详细度
  • force_new：强制新实例（测试/隔离）

get_instance / reset_instance

• 类型：classmethod
• 状态：beta
• 描述： 读当前单例 / 清空单例（测试用）

get_table / register_table

• 状态：beta
• 描述： 按逻辑表名取已注册 Model 实例；从「表目录」（含 schema.py/model.py）注册

get_physical_table_name

• 状态：beta
• 描述： 逻辑名 → 物理名（如 PostgreSQL schema.table）

normalize_delist_date

DataManager.normalize_delist_date(raw) -> str | None（staticmethod）

• 状态：beta
• 描述： 将源数据占位退市日（0 / 0.0）视为未退市；跨模块入口

register_calendar_real_world_fetcher

DataManager.register_calendar_real_world_fetcher(fetcher) -> None（staticmethod）

• 状态：beta
• 描述： 注入日历 real-world 网络探测（Callable[[], tuple[date, provider] | None]）。通常由 DataSourceManager 在导入/构造时注册；DM 不依赖 DS

attach_data_service

DataManager.attach_data_service() -> DataService

• 状态：beta
• 描述： 绑定 / 重建领域 DataService（worker 子进程、DuckDB pool resume）。跨模块勿 deep-import core.data_services.DataService

ensure_duckdb_pool_holder_resolver

DataManager.ensure_duckdb_pool_holder_resolver() -> None（classmethod）

• 状态：beta
• 描述： 向 infra.db 注册 holder 解析（单例 / 可选新建），避免 infra import DataManager。包导入与构造时自动调用

ensure_restored_after_worker_pool

DataManager.ensure_restored_after_worker_pool(data_mgr=None) -> DataManager | Any（classmethod）

• 状态：beta
• 描述： ProcessPool / 中断后恢复主进程 DM + DB 的应用层入口

bind_as_default_instance

DataManager.bind_as_default_instance() -> None

• 状态：beta
• 描述： DuckDB pool resume 后把本实例挂回进程单例（供 infra duck-type 调用）

领域服务（属性）

举例：

from core.modules.data_manager import DataManager

dm = DataManager(is_verbose=True)
klines = dm.stock.kline.load("000001.SZ", term="daily")
# 每行形状见 strategy/docs/PRICE_LAYERS.md §2：
# 顶层 OHLC = qfq；row["raw"] / row["hfq"]（raw×F）/ row["adj_factor"]
# load(..., adjust=...) 已忽略；未复权请用 load_raw
open_dates = dm.calendar.load_open_dates("20240101", "20241231")

sample_universe（嵌套）

DataManager.sample_universe（namespace；实现类不从包根导出）

• 状态：beta
• 描述： 开发样本股票宇宙。由 data.json 的 use_sample_stock_list 开关 + core/dev/sample_stock_list/stratified_N.csv 决定系统内保留哪些股票。抓取侧与写库过滤消费此宇宙；ListService 读路径信任 DB。
• 常用：
  • count() / ids() / is_active() / csv_path(count) / directory() / invalidate_cache()
  • slice_stock_list(rows) / slice_stock_list_in_dependencies(deps)
  • filter_records(records, schema) / filter_paired(main, raw) / prune(data_manager)
  • DEFAULT_COUNT（分层 CSV 默认规模）

举例：

from core.modules.data_manager import DataManager

rows = DataManager.sample_universe.slice_stock_list(full_list)
ids = DataManager.sample_universe.ids()

contracts