market_sync/docs/ARCHITECTURE.md

Market Data Sync — 架构说明

A 股市场数据定时同步框架。多个数据源 → 交易日感知调度器 → MySQL。

1. 目录结构

market_data_sync/
├── app/                          # 业务代码（唯一 Python 包）
│   ├── __init__.py
│   ├── core/                     # 核心抽象层（与业务无关的基础设施）
│   │   ├── config.py             # 统一配置（pydantic-settings + .env）
│   │   ├── datasource/           # 数据源抽象
│   │   │   ├── base.py           # DataSource / FetchResult / registry
│   │   │   └── registry.py       # 健康监控 + 自动降级 + seed
│   │   ├── sync/                 # 同步任务抽象
│   │   │   ├── base.py           # SyncTask 基类（自动状态机）
│   │   │   └── registry.py       # dataset_registry 状态机封装
│   │   ├── scheduler/            # 轻量调度器
│   │   │   └── scheduler.py      # 交易日感知 + config 表持久化
│   │   ├── db/                   # 数据库访问层
│   │   │   ├── connection.py     # thread-local pymysql
│   │   │   ├── schema.py         # 13 张表 DDL
│   │   │   └── ops.py            # CRUD + 批量 upsert
│   │   └── utils/
│   │       └── logging.py        # RotatingFileHandler
│   ├── sources/                  # 数据源实现（5 个）
│   │   ├── sina.py               # 新浪（免费 K 线）
│   │   ├── baostock.py           # Baostock（基础信息+行业）
│   │   ├── mairui.py             # 麦蕊智数（需 MAIRUI_LICENCE）
│   │   ├── akshare.py            # akshare/东方财富（5min+资金流）
│   │   └── xueqiu.py             # 雪球（股本快照）
│   ├── tasks/                    # 同步任务实现（8 个）
│   │   ├── __init__.py           # TASKS 注册表 + get_task()
│   │   ├── task_stocks_basic.py
│   │   ├── task_kline_daily.py
│   │   ├── task_kline_index.py
│   │   ├── task_kline_5min.py
│   │   ├── task_moneyflow.py
│   │   ├── task_industry_sector.py
│   │   ├── task_share_snapshot.py
│   │   └── task_market_regime.py
│   ├── api/                      # FastAPI 管理接口（端口 8100）
│   │   ├── main.py               # 入口 + 生命周期钩子
│   │   └── routes/
│   │       ├── health.py
│   │       ├── sync.py           # 手动触发 / 状态查询
│   │       ├── schedule.py       # 计划任务
│   │       └── datasources.py    # 数据源健康度
│   └── entrypoints/              # 进程入口
│       ├── cli.py                # CLI（python -m app.entrypoints.cli ...）
│       └── worker.py             # 纯调度模式（python -m app.entrypoints.worker）
├── bin/                          # 一次性脚本
│   └── runall_once.py            # 一键串行跑全部任务
├── scripts/                      # 旧 shell 脚本（保留作 fallback）
├── tests/                        # 单元测试（pytest）
├── docs/                         # 文档
├── data/                         # 临时数据（一般为空）
├── logs/                         # 运行日志（RotatingFileHandler）
├── .env / .env.example           # 配置
├── requirements.txt
├── README.md
└── start.sh                      # 统一启动入口

2. 模块依赖

            ┌────────────────────────────────────────┐
            │            entrypoints/cli             │  手动/CLI
            │            entrypoints/worker          │  调度模式
            │                  │                     │
            │                  ▼                     │
            │  ┌─────────────────────────────┐       │
            │  │        api (FastAPI)        │       │  HTTP 管理
            │  └──────────────┬──────────────┘       │
            │                 │                      │
            ▼                 ▼                      │
       ┌──────────┐    ┌─────────────┐               │
       │  tasks   │◄───│   core      │               │
       │  (8 个)  │    │  ├ config   │               │
       └────┬─────┘    │  ├ db       │               │
            │          │  ├ sync     │               │
            │          │  ├ scheduler│               │
            │          │  └ datasource│              │
            ▼          └──────┬──────┘               │
       ┌──────────┐           │                      │
       │ sources  │◄──────────┘                      │
       │  (5 个)  │                                  │
       └──────────┘                                  │
            │                                         │
            ▼                                         │
        数据源外部 API                                 │

依赖方向单向：上层 → 下层。core 是稳定抽象，sources / tasks 是可插拔实现。

3. 核心抽象

3.1 DataSource (app/core/datasource/base.py)

class DataSource:
    key: str                       # "datasource_xxx" 全局唯一
    name: str                      # 显示名
    provides: list[str]            # 能力列表 ["kline_daily", "moneyflow", ...]
    requires_credential: bool

    def is_available(self) -> tuple[bool, str]: ...
    def health_check(self) -> dict: ...
    def fetch_xxx(self, ...) -> pd.DataFrame: ...

• registry 是全局单例 dict[str, DataSource]
• 启动时 build_default_registry() 把 5 个源注册进去
• pick_source(capability) 按 provides 找最匹配的源

3.2 SyncTask (app/core/sync/base.py)

class SyncTask:
    dataset_id: str                # 子类必须设置

    def run(self, *, trigger_source="manual", **kwargs) -> dict:
        mark_sync_running(...)
        try:
            result = self._run(...)
            mark_sync_success/failed(...)
        except Exception:
            mark_sync_failed(...)

子类只需实现 _run()，状态机会自动包装。

3.3 Scheduler (app/core/scheduler/scheduler.py)

• 5s tick 扫描 config 表
• 读取 trading_calendar_holidays 决定是否交易日
• 时间匹配（HH:MM 精确触发）或 interval（间隔秒数）触发
• 启动时 recover_interrupted_syncs() 把卡死的 running 任务改 failed

4. 数据流（以 kline_daily 为例）

scheduler tick
  └─ 时间到 (16:00)
      └─ ScheduleJob.run()
          └─ tasks.get_task("kline_daily")()
              └─ SyncKlineDaily.run()
                  ├─ mark_sync_running(...)            ← 状态机
                  ├─ _ensure_health_checked()          ← 拉一次数据源健康
                  ├─ pick_source("kline_daily")        ← 选雪球(主) > 新浪 > baostock
                  ├─ 多线程 fetch (10 workers)         ← XUEQIU_TOKEN 限速
                  ├─ 批量 upsert (CHUNK=500)           ← pymysql executemany
                  └─ mark_sync_success(...)            ← 状态机

5. 添加新数据源

1. 在 app/sources/ 新建 my_source.py
2. 继承 DataSource，设置 key / name / provides / requires_credential
3. 实现 is_available() / health_check() / 业务方法
4. 在 app/core/datasource/registry.py 的 build_default_registry() 注册

6. 添加新同步任务

1. 在 app/tasks/ 新建 task_xxx.py
2. 继承 SyncTask，设置 dataset_id
3. 实现 _run()，返回 {"status": "ok", "message": "..."} 或 {"status": "warning/blocked", ...}
4. 在 app/tasks/__init__.py 的 TASKS 字典中加一行
5. （可选）在 app/core/sync/registry.py 的 SYNC_DEFINITIONS 加条目

7. 数据库表

13 张表在 app/core/db/schema.py：

表	用途
stocks	股票基础信息
kline_stock	日 K 线
kline_index	指数日 K
kline_5min	5min K 线
moneyflow	资金流
industry / sectors / stock_sector_map	行业映射
share	股本快照
market_regime_daily	市场情绪
dataset_registry	同步任务状态机
config	通用配置（调度、节假日、token 等）
indices	指数字典

8. 配置（.env）

变量	必填	说明
MYSQL_HOST/PORT/USER/PASSWORD/DATABASE	✓	主库
MYSQL_DATABASE_BUSINESS		业务库（默认 grid_seeker_model_base）
XUEQIU_TOKEN		雪球 token（K线主源 + 股本）
MAIRUI_LICENCE		麦蕊智数 licence
HOLIDAYS_LIST		A 股休市日（逗号分隔 YYYY-MM-DD）
SCHEDULER_AUTO_SEED		启动时 seed 默认计划任务
LOG_DIR		日志目录（默认 logs）
LOG_LEVEL		日志级别（默认 INFO）