# data **Repository Path**: quant-seminar/data ## Basic Information - **Project Name**: data - **Description**: No description available - **Primary Language**: Unknown - **License**: Not specified - **Default Branch**: dev - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 1 - **Created**: 2026-04-04 - **Last Updated**: 2026-07-12 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # data `data` 是 Seminar 项目的数据基座服务，负责日频股票因子、ST 状态和指数成分权重的采集、存储与查询。服务对外提供 FastAPI 接口，底层使用 Tushare 获取数据、DolphinDB 持久化，并支持通过结构化 JSON 动态构造派生因子。 线上部署地址： - 服务根地址：https://data.realhuhu.com/ - Swagger UI：`/docs` - OpenAPI JSON：`/openapi.json` ## 功能概览 - 定时更新：生产环境每天 19:00 更新普通股票日频因子。 - 独立更新：普通日频因子、ST 状态和指数权重分别由独立 worker 维护。 - 数据查询：支持显式股票代码或指数成分股票池，并可按 ST 状态和自定义条件过滤。 - 指数权重：提供元数据、区间查询和后台增量更新接口。 - 派生因子：支持用 `Derivative` JSON 构造时序、截面、TA-Lib 和嵌套表达式。 - 依赖排序：命名派生因子会根据字段引用自动拓扑排序，调用方无需手动按计算顺序排列 JSON key。 ## 服务结构 ```text data |-- main.py # FastAPI 入口和定时任务 |-- config.py # 环境变量配置 |-- core | |-- database/session.py # DolphinDB 连接、建表和统一写入 | |-- routers/stock_daily | | |-- view.py # 日频因子 HTTP 路由与 OpenAPI 声明 | | |-- schema.py # 请求与响应模型 | | `-- func.py # 查询、筛选和后台更新逻辑 | |-- routers/index_weight | | |-- view.py # 指数权重 HTTP 路由与 OpenAPI 声明 | | |-- schema.py # 请求与响应模型 | | `-- func.py # 数据查询和后台更新逻辑 | |-- workers/stock_daily/base.py # 普通日频因子 worker 框架 | |-- workers/stock_daily/volume_price.py | |-- workers/stock_daily/financial.py | |-- workers/stock_st.py # 稀疏 ST 状态 worker | |-- workers/index_weight.py # 指数权重 worker | |-- operators # 派生因子 DSL 和依赖排序 | `-- utils # Tushare、查询、限流、日志工具 |-- tests # 单元测试和 opt-in 实库测试 `-- pyproject.toml ``` ## 数据模型 DolphinDB 中包含两张分区表。股票日频数据使用统一长表： | 字段 | 类型 | 说明 | | --- | --- | --- | | `time` | `TIMESTAMP` | 数据日期或公告日期 | | `code` | `SYMBOL` | 股票代码，例如 `000001.SZ` | | `factor` | `SYMBOL` | 因子名，例如 `close_hfq` | | `value` | `DOUBLE` | 因子值 | 查询时服务会根据 `time/code/factor` 过滤数据，并 pivot 成宽表用于派生因子计算。 `derive_factors` 中每个命名因子都会作为宽表列写回，后续因子可以通过字段名引用它。 `st` 也存储在该长表中，但采用稀疏存储：只有某只股票某日为 ST 时才写入 `value=1`，非 ST 不写行。查询宽表会把缺失的 `st` 补为 `0`；其他稀疏因子按股票向前填充。 `industry` 不写入 DolphinDB。查询需要该字段时，服务根据 `stock_basic.industry` 和 `INDUSTRY_TO_SECTOR` 实时映射行业大类，未匹配行业归为“工业”。它既可以直接放入 `factors` 输出，也可以作为分类控制变量供 `multiary.neutralize_by` 使用；若查询只需要 `industry`，服务使用 `close` 建立有效的 `time/code` 行。 指数成分权重表结构如下： | 字段 | 类型 | 说明 | | --- | --- | --- | | `time` | `TIMESTAMP` | 权重快照日期 | | `index` | `SYMBOL` | 指数代码，例如 `000300.SH` | | `code` | `SYMBOL` | 成分股代码 | | `weight` | `DOUBLE` | 该快照中的成分权重 | ## 配置 服务会加载当前目录和上级目录中的 `.env`： ```env PROD=true DOLPHIN_HOST=dolphindb DOLPHIN_PORT=8848 DOLPHIN_USERNAME=admin DOLPHIN_PASSWORD=123456 DOLPHIN_DAILY_DB=... DOLPHIN_DAILY_STOCK_TB=... DOLPHIN_DAILY_INDEX_WEIGHT_TB=indexWeight INDEX_CODES=000016.SH,000300.SH,000905.SH,000852.SH TUSHARE_TOKEN=... ``` 在主项目中可通过 `docker-compose.yml` 启动，默认映射： | 服务 | 容器端口 | 主机端口 | | --- | --- | --- | | `seminar-data` | `8000` | `8800` | ## 启动 本地开发： ```bash uv sync uvicorn main:app --host 127.0.0.1 --port 8000 --reload ``` 通过主项目 Docker Compose： ```bash docker compose up --build seminar-data ``` ## API ### GET `/stock/daily/metadata` 返回当前服务支持的股票代码、基础因子列表和因子构造说明。 响应结构： ```json { "code": 100, "msg": "OK", "data": { "codes": ["000001.SZ"], "index_codes": ["000016.SH", "000300.SH", "000905.SH", "000852.SH"], "factors": ["open", "close_hfq"], "update_handlers": ["daily", "daily_hfq", "daily_qfq", "daily_basic", "balance_sheet", "income", "cashflow", "fina_indicator", "stock_st"], "instruct": "..." } } ``` 其中 `instruct` 是给调用方或大模型使用的派生因子构造规范。 ### POST `/stock/daily/query` 查询日频因子数据，返回 parquet 二进制文件。 请求体： ```json { "start_date": "2024-01-01", "end_date": "2024-12-31", "codes": ["000001.SZ", "600000.SH"], "drop_st": false, "auto_mask": true, "factors": ["close_hfq", "vol"], "derive_factors": { "CLOSE_MA20": { "type": "TS", "op": "unary.rolling", "fields": {"col": "close_hfq"}, "params": {"window": 20, "agg": "mean"} } }, "filter_conditions": { "keep_non_st": { "type": "TS", "op": "binary.eq", "fields": {"left": "st", "right": 0}, "params": {} } } } ``` 字段说明： | 字段 | 说明 | | --- | --- | | `start_date` | 开始日期，格式 `YYYY-MM-DD` | | `end_date` | 结束日期，格式 `YYYY-MM-DD` | | `codes` | 显式股票代码列表，与 `index_code` 必须且只能提供一个 | | `index_code` | 使用该指数的历史成分作为每日股票池，与 `codes` 必须且只能提供一个 | | `drop_st` | 是否排除 ST 股票，默认 `false` | | `auto_mask` | 是否自动应用最终 mask，默认 `true`；为 `false` 时保留全部已查询行并返回 bool 列 `mask` | | `factors` | 需要查询并输出的基础因子列表 | | `derive_factors` | 用户自定义派生因子，类型为 `dict[str, Derivative]` | | `filter_conditions` | 返回 bool 或数值的保留条件，类型为 `dict[str, Derivative]` | 使用 `index_code` 时，服务先查询区间内所有权重快照及开始日期之前最近一期快照，并获取这些快照中出现过的股票。每个行情日使用“当日及之前最近一期快照”判断成分资格，因此调仓前后的股票池会自动变化。 查询阶段构造可选的 `cross_section_mask`：指数成员条件与 `drop_st=true` 时的 `st == 0` 条件按 AND 合并。该 mask 只限制 CS 因子的计算截面，并在最后参与选行；TS 因子仍在已查询到的完整股票时序上计算。 最终 mask 由 `cross_section_mask` 与全部 `filter_conditions` 按 AND 合并。`auto_mask=true` 时只返回 mask 为 `true` 的行；`auto_mask=false` 时不删除任何已查询行，而是在结果末尾追加 bool 类型的 `mask` 列。此时 `mask=true` 表示该行在默认自动筛选模式下会被保留；若没有指数、ST 或条件限制，`mask` 全为 `true`。 `filter_conditions` 与 `derive_factors` 合并到同一个依赖图，并由 `compute_derives` 一次完成计算，因此条件可以引用命名派生因子。所有条件按 AND 合并：bool 值直接判断，数值以非零为 `true`，空值视为 `false`。条件过滤发生在 CS 因子计算之后，不会用过滤后的行重新计算截面；条件列仅用于内部筛选，不出现在返回结果中。 返回 parquet 的前两列固定为 `time`、`code`，随后为输出基础因子和命名派生因子；结果行按 `code/time` 排序。 ### 派生因子计算顺序 `derive_factors` 是一个对象，但服务不会直接按对象遍历顺序计算。`data` 会分析每个 `Derivative.fields` 中对其它命名派生因子的引用，并执行拓扑排序： ```json { "FINAL_FACTOR": { "type": "TS", "op": "binary.mul", "fields": { "left": "MOM_20D", "right": "LOW_VOL" }, "params": {} }, "MOM_20D": { "type": "TS", "op": "unary.pct_change", "fields": { "col": "close" }, "params": { "periods": 20 } }, "LOW_VOL": { "type": "TS", "op": "binary.div", "fields": { "left": 1, "right": { "type": "TS", "op": "unary.rolling", "fields": { "col": "pct_chg" }, "params": { "window": 20, "agg": "std" } } }, "params": {} } } ``` 即使 `FINAL_FACTOR` 写在最前面，服务也会先计算 `MOM_20D` 和 `LOW_VOL`，再计算 `FINAL_FACTOR`。如果出现 `A -> B -> A` 这样的循环依赖，会返回清晰的循环依赖错误。 独立因子之间会保留输入顺序，因此多因子策略合并多个已发布因子时，最终因子的选择顺序仍可由上游 `factor_names` 控制。 ### POST `/stock/daily/update` 启动普通股票日频和 ST 后台更新任务。可通过查询参数 `handlers` 传入逗号分隔的更新目标： ```text POST /stock/daily/update?handlers=daily,daily_basic,stock_st ``` 普通日频 handler 可选值为 `daily`、`daily_hfq`、`daily_qfq`、`daily_basic`、`balance_sheet`、`income`、`cashflow`、`fina_indicator`；`stock_st` 表示运行独立的 `StockSTWorker`。名称会去重，空项或未知名称返回 400。省略 `handlers` 时默认运行全部普通 handler 和 `stock_st`。 服务通过进程内锁保证同一时间只有一个股票日频后台更新任务。普通 handler 作为一组交给 `DailyStockWorker.run()`，随后运行 `StockSTWorker`；普通更新失败不会阻止 ST 更新。 响应示例： ```json { "code": 100, "msg": "已启动后台更新任务", "data": { "handlers": ["daily", "daily_basic", "stock_st"] } } ``` ### GET `/index/weight/metadata` 返回当前配置的指数代码列表，默认包括上证50、沪深300、中证500和中证1000。 ### POST `/index/weight/query` 查询一个或多个指数在日期区间内的原始权重快照，返回列为 `time/index/code/weight` 的 parquet： ```json { "start_date": "2019-01-01", "end_date": "2019-12-31", "index_codes": ["000300.SH"] } ``` ### POST `/index/weight/update` 启动 `IndexWeightWorker` 后台增量更新。接口通过进程内锁拒绝同一进程中的重复任务。 ## 派生因子 DSL `derive_factors` 使用结构化 JSON 描述计算图。核心模型如下： ```python Col = Union[str, "Derivative", int, float, bool] class Derivative(BaseModel): type: Literal["TS", "CS"] = "TS" op: str fields: dict params: dict = {} ``` `type` 决定分组方式： | 类型 | 分组 | 适用场景 | | --- | --- | --- | | `TS` | 按 `code` 分组 | 均线、滚动统计、滞后项、技术指标 | | `CS` | 按 `time` 分组 | 截面排名、z-score、残差、中性化 | `op` 格式为 `group.operation`，例如： - `unary.rolling` - `binary.div` - `multiary.add` - `talib.RSI` - `nullary.weekday` ### 示例：20 日均线 ```json { "MA20": { "type": "TS", "op": "unary.rolling", "fields": { "col": "close_hfq" }, "params": { "window": 20, "agg": "mean" } } } ``` ### 示例：5 日收益率 ```json { "RET_5D": { "type": "TS", "op": "unary.pct_change", "fields": { "col": "close_hfq" }, "params": { "periods": 5 } } } ``` ### 示例：截面百分位排名 ```json { "PB_RANK": { "type": "CS", "op": "unary.rank_pct", "fields": { "col": "pb" }, "params": {} } } ``` ### 示例：嵌套因子 先计算每日成交量截面排名，再对排名做 5 日均值： ```json { "VOL_RANK_MA5": { "type": "TS", "op": "unary.rolling", "fields": { "col": { "type": "CS", "op": "unary.rank_pct", "fields": { "col": "vol" }, "params": {} } }, "params": { "window": 5, "agg": "mean" } } } ``` 完整 DSL 说明以 `/stock/daily/metadata` 返回的 `instruct` 为准。 ## 数据更新流程 普通日频因子由 `DailyStockWorker` 调度： 1. 根据 DolphinDB 中已有数据获取每个 `code/factor` 的最近日期。 2. 每个 handler 根据最近日期增量调用 Tushare。 3. 将宽表结果转换为 `time/code/factor/value` 长表。 4. 分批写入 DolphinDB。 当前注册的数据来源包括： - `volume_price.py`：基础日线、前复权、后复权、每日指标。 - `financial.py`：资产负债表、利润表、现金流量表、财务指标，以及季度和 TTM 字段。 另外两个 worker 独立运行，不注册到 `DailyStockWorker`： - `StockSTWorker`：按交易日调用 `stock_st`，只写入当日 ST 股票的 `st=1` 记录；非 ST 不写入。 - `IndexWeightWorker`：按配置的指数分别读取数据库回执，使用 `end_date` 从当前日期向历史分页，直至覆盖最新已存快照。 生产环境的 19:00 定时任务目前只运行 `stock_daily_worker`。指数权重可通过 `/index/weight/update` 更新；ST 数据可直接运行 `stock_st_worker.run()`。 ## 注意事项 - 价格类技术指标优先使用后复权字段，例如 `close_hfq`、`open_hfq`、`high_hfq`、`low_hfq`。 - 利润表和现金流量表字段优先使用 TTM 字段，例如 `revenue_ttm`。 - 使用截面操作时必须显式声明 `"type": "CS"`。 - `Derivative` 只能出现在 `fields` 中，`params` 中应只放常量、列表或普通字典。 - `factors`、`derive_factors`、`filter_conditions` 的名称不能冲突；自定义名称也不能与原始因子或 `code/time` 同名。 - 如果一个中间表达式只服务于最终因子，优先使用嵌套 `Derivative`，减少不必要的命名列。 ## 测试 运行完整单元测试： ```bash uv run python -m unittest discover -s tests ``` 测试覆盖派生因子解析与依赖排序、截面中性化、数据库建表与写入、指数权重和 ST worker，以及股票日频查询的股票池、ST 和条件 mask 逻辑。 - JSON key 顺序与计算顺序不一致时仍能正确计算。 - 独立依赖保留输入顺序。 - 截面因子依赖时序因子。 - 嵌套表达式与命名依赖混用。 - 循环依赖和缺失基础列返回明确错误。 2019 年沪深 300 实库集成测试默认跳过。确认本机 DolphinDB 已包含股票日线、ST 和指数权重数据后，使用 PowerShell 显式运行： ```powershell $env:RUN_REAL_DATA_TESTS = "1" uv run python -m unittest tests.test_stock_daily_2019_real -v ``` 该用例使用 `2019-06-27` 至 `2019-07-02` 的真实数据，覆盖沪深300在 `2019-06-28` 调入、调出各 19 只股票，验证 ST 排除、条件过滤、截面 mask，以及 CS 因子先计算后过滤的执行顺序。预期值直接读取 DolphinDB 原始表后独立计算，再与接口 parquet 逐行比较。