# fdata

**Repository Path**: carllee34/fdata

## Basic Information

- **Project Name**: fdata
- **Description**: 一个金融数据处理项目
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-07-07
- **Last Updated**: 2026-10-08

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# FData - 金融数据同步与处理系统

基于 **QuestDB** 主库的金融数据同步与处理系统，整合 **baostock** 和 **akshare** 双数据源，通过 **RabbitMQ + Apache Flink** 构建实时/准实时流处理层，支持海量时序数据的高效采集、标准化、衍生计算与存储，并保留 **PostgreSQL + pgvector** 作为未来向量检索扩展。

---

## 核心特性

- **双数据源互补**：baostock（A股历史数据，稳定可靠）+ akshare（多市场/实时/另类数据，覆盖面广）
- **QuestDB 时序主库**：日线按年分区，分钟线按月分区，支持高吞吐写入与金融时序查询
- **RabbitMQ 消息总线**：统一接入层，解耦数据采集与流处理
- **Flink 流处理层**：K 线标准化、去重、衍生指标、周/月线实时重采样、涨跌停价衍生、CEP 形态检测
- **完整数据覆盖**：A股日线/财务/估值/行业分类/指数成分股/资金面/宏观/期货/基金等 114 张表
- **增量同步引擎**：日终自动增量更新，支持断点续传与缺口自愈
- **pgvector 向量检索预留**：K线形态相似性搜索、因子聚类、文本语义检索（Phase 9）

---

## 快速开始

### 1. 前置条件

- Docker & Docker Compose
- Java 17 + Maven 3.8+（构建 Flink 作业）
- conda 环境 `x_quant`（Python 3.12+）

### 2. 启动全栈服务

```bash
# QuestDB 数据通过 bind-mount 持久化在 <项目根>/data/questdb（无需创建命名卷）

cd docker
# 首次启动建议先构建自定义 RabbitMQ 镜像
docker compose build rabbitmq
# 启动 QuestDB + RabbitMQ + Flink + Sink Worker + Collector（核心服务）
# collector 为采集调度容器（首次或代码变更需 docker compose build collector）
docker compose up -d
```

访问地址：
- QuestDB Web Console: http://localhost:9000
- QuestDB PG wire: `localhost:8812`
- RabbitMQ Management: http://localhost:15672 （默认 `fdata/fdata_pass`）
- Flink UI: http://localhost:8081

### 3. 构建 Flink 作业

```bash
cd src/flink
mvn clean package -DskipTests
# Flink 作业统一通过脚本提交（compose 中无 flink-submit 容器）
bash scripts/deploy-flink-jobs.sh
```

### 4. 数据源采集（日常调度在 collector 容器；按需采集可本地执行）

日常定时采集（日终 K 线/复权/ETF、除权除息、晨检、Flink 看门狗）由 `fdata-collector`
容器内 APScheduler 自动触发（2026-09-29 起，宿主机无需 crontab）：

```bash
docker logs -f fdata-collector                       # 查看调度器与任务执行日志
docker compose exec collector python sync/daily_market.py   # 容器内手动补跑
```

常驻/批量采集（run_ingest 进程管理器）与手动补数仍可在本机直接运行
（自行决策采集节奏，与 collector 经 BaostockGate 共享锁互斥）：

```bash
cd src
# 方式一：进程管理器统一启动（并发限制 + 崩溃自动重启）
PYTHONPATH=$(pwd) INGEST_SOURCE=baostock python pipeline/run_ingest.py

# 方式二：离线同步脚本（宿主机手动执行）
conda run -n x_quant python sync/run.py          # 六域增量（basics→kline→adjust→dividend→financial→futures）
conda run -n x_quant python sync/daily_market.py # K线+复权+ETF（每日更新 API，3 请求/全市场）
```

> 说明：采集容器（ingest-*）已从 docker-compose.yml 移除（2026-08）；2026-09-29 起日常定时
> 采集由 collector 容器统一调度（调度器只在交易日定时窗口触发，不会 24/7 自动采集）。
> 数据源为 baostock + akshare 双源（tqsdk/tushare 通道已于 2026-08-11 移除）。

### 5. 初始化基础数据（批量离线同步，可选）

如果你更习惯离线批量方式，仍可执行（`sync/` 目录，详见 `sync/README.md`）：

```bash
# 同步交易日历 + 全市场证券列表（akshare 源）
conda run -n x_quant python sync/basics_akshare.py

# 全量同步历史日K线（baostock 逐只，约需数小时）
conda run -n x_quant python sync/kline.py --start 1990-12-19

# 周月线（从日K聚合生成，零 API）/ 复权因子 / 除权除息 / 财务摘要
conda run -n x_quant python scripts/generate_wm_from_daily.py --mode full
conda run -n x_quant python sync/adjust_factor.py
conda run -n x_quant python sync/dividend_akshare.py
conda run -n x_quant python sync/financial_akshare.py

# akshare 扩展数据缺口补全
conda run -n x_quant python scripts/backfill_akshare_tables.py
```
> 旧的 `scripts/sync_kline.py`/`sync_adjust_factors.py`/`sync_dividend.py`/`sync_financial.py`/`daily_sync.py` 已归档至 `scripts/archive/`（2026-08-19），勿再使用。

### 6. 日终增量同步

日终任务已由 `fdata-collector` 容器自动调度（周一至周五 18:00 `daily_market --republish 2`、
18:30 `run.py --domain dividend`，任务表见 `src/pipeline/scheduler.py`），宿主机无需 crontab。
手动执行（宿主机或容器内均可）：

```bash
conda run -n x_quant python sync/daily_market.py   # K线+复权+ETF
conda run -n x_quant python sync/run.py             # 五域全量增量
```

### 7. 验证数据流

```bash
python3 tests/test_e2e_pipeline.py
```

---

## 数据流架构

```
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  baostock /     │     │   RabbitMQ      │     │  Apache Flink   │
│  akshare SDK    │────▶│   fdata.raw.*   │────▶│  Normalization  │
│  (Python)       │     │   消息总线       │     │  Derived Ind.   │
└─────────────────┘     └─────────────────┘     └─────────────────┘
                              │                         │
                              ▼                         ▼
                       q.kline.daily            q.normalized.kline.daily
                       q.financial.*            q.derived.indicators
                       q.akshare.*              q.metadata.*
```

- **Producer**：`src/pipeline/` 下的 Python ingestion services。
- **Broker**：RabbitMQ，队列与绑定定义在 `docker/rabbitmq/definitions.json`。
- **Stream Processor**：Flink 1.20，作业入口 `src/flink/src/main/java/com/fdata/flink/FdataFlinkJobs.java`。
- **Sink**：标准化 K 线通过 `QuestDBSink` 批量写入 QuestDB `kline_daily`；元数据通过对应 binder 写入 `symbols` / `trade_calendar`；其余作业结果写回 RabbitMQ 供下游消费。

---

## 项目结构

```
fdata/
├── docker/                          # Docker 部署配置
│   ├── docker-compose.yml           # 核心数据管道（QuestDB/RabbitMQ/Flink/Sink Worker/Collector 采集调度）
│   └── rabbitmq/                    # RabbitMQ 自定义镜像与 definitions
│
├── src/                             # 源码
│   ├── config/                      # 配置管理
│   │   └── database.py              # 数据库连接 / QuestDB 连接
│   │
│   ├── models/                      # ORM 模型（22 个模型类，schema 自文档化）
│   │   ├── symbol.py                # symbols / trade_calendar / adjust_factors
│   │   ├── kline.py                 # kline_daily / minute / weekly / monthly
│   │   ├── financial.py             # financial_quarterly / express / forecast
│   │   ├── akshare_ext.py           # fund_flow / margin / longhubang / block_trade
│   │   ├── shareholder.py           # shareholder_holding / restricted_release
│   │   ├── macro.py                 # macro_indicators
│   │   └── embedding.py             # kline/factor/text embeddings (pgvector)
│   │
│   ├── data_sources/                # 数据源封装（仅 baostock + akshare 双源）
│   │   ├── baostock_source.py       # baostock 25+ 接口全封装（含 BaostockGate 全局互斥）
│   │   ├── akshare_source.py        # akshare 核心接口封装（限速/熔断/重试）
│   │
│   ├── pipeline/                    # Python Ingestion Services
│   │   ├── base_ingest.py           # 抽象基类
│   │   ├── message.py               # 统一消息格式
│   │   ├── rmq.py                   # RabbitMQ 发布者
│   │   ├── ingest_baostock_*.py     # baostock ingestion 服务
│   │   ├── ingest_akshare_*.py      # akshare ingestion 服务
│   │   ├── scheduler.py             # collector 容器内调度器（APScheduler，原宿主机 crontab 任务）
│   │   └── Dockerfile.sink-worker / Dockerfile.collector  # Sink Worker / Collector（采集调度）镜像
│   │
│   ├── flink/                       # Flink 流处理作业（Java/Maven）
│   │   ├── pom.xml
│   │   ├── README.md
│   │   └── src/main/java/com/fdata/flink/
│   │       ├── FdataFlinkJobs.java  # 统一入口
│   │       ├── jobs/                # 各 Flink 作业
│   │       ├── model/               # 消息 POJO
│   │       └── sink/                # RabbitMQSink / QuestDBSink
│   │
│   ├── sync/                        # 空目录（后续 Phase 扩展；日常同步脚本在根目录 sync/）
│   ├── processors/                  # 空目录（待扩展）
│   ├── embeddings/                  # 空目录（待扩展）
│   └── utils/                       # 工具（logger/retry/questdb_client/questdb_bulk/sync_helpers/baostock_gate/rate_limiter/circuit_breaker）
│
├── scripts/                         # 可执行脚本
│   ├── backfill_*.py                # akshare 扩展域缺口补全工具集（backfill_daily_gaps.py 为全域日常入口）
│   ├── generate_wm_from_daily.py    # 周月线聚合生成（零 API）
│   ├── generate_limit_price*.py     # 涨跌停价衍生
│   ├── data_freshness_check.py      # 数据新鲜度晨检
│   ├── flink_watchdog.sh / flink_queue_drain.sh  # Flink 看门狗 / 队列滞留放行
│   ├── cleanup_dirty_data.py        # 通用脏数据清理应急工具
│   ├── deploy-flink-jobs.sh         # Flink 作业提交脚本
│   └── archive/                     # 已归档脚本（2026-08-19 归档 18 个 + 2026-09-20 归档 8 个）
│
├── tests/                           # 测试（管道映射/transform 单测 + 端到端）
├── notebooks/                       # Jupyter 笔记本
├── data/                            # QuestDB 数据 bind-mount（data/questdb）+ 备份/Flink checkpoint 卷
├── logs/                            # 日志输出目录
├── .env.example                     # 环境变量模板
└── requirements.txt                 # Python 依赖
```

---

## 数据库 Schema

> 当前生产写入统一走 **QuestDB**；以下表名与 `src/models/` 中的 ORM 模型对应。

### 元数据层
| 表 | 说明 | 数据量 |
|---|---|---|
| `symbols` | 资产主表（股票/指数/基金/期货等） | ~7,800 |
| `trade_calendar` | 交易日历 | ~11,400 |
| `adjust_factors` | 复权因子 | 动态 |

### 行情数据层（QuestDB 分区表）
| 表 | 说明 | 分区策略 |
|---|---|---|
| `kline_daily` | 日K线（含PE/PB/PS/PCF估值） | 按年分区（1990-2030） |
| `kline_minute` | 分钟K线 | 按月分区 |
| `kline_weekly` | 周K线 | - |
| `kline_monthly` | 月K线 | - |

### 财务数据层
| 表 | 说明 | 数据来源 |
|---|---|---|
| `financial_quarterly` | 6类季频财务指标统一入口 | baostock |
| `performance_express` | 业绩快报 | baostock |
| `performance_forecast` | 业绩预告 | baostock |
| `dividend_events` | 除权除息 | baostock |

### akshare 扩展层
| 表 | 说明 |
|---|---|
| `fund_flow_daily` | 个股资金流向 |
| `margin_trading` | 融资融券 |
| `longhubang` | 龙虎榜 |
| `block_trade` | 大宗交易 |
| `shareholder_holding` | 股东持股变动 |
| `restricted_release` | 解禁股 |
| `macro_indicators` | 宏观经济指标 |

### pgvector 向量层（未来扩展）
| 表 | 维度 | 用途 |
|---|---|---|
| `kline_embeddings` | 768 | K线形态相似性搜索 |
| `factor_embeddings` | 256 | 因子聚类/相似股搜索 |
| `text_embeddings` | 1536 | 新闻/研报语义检索 |

---

## 数据源能力对比

| 数据类型 | baostock | akshare | 推荐主源 |
|---|---|---|---|
| A股历史日K线（个股 2005 年起、指数 1990 年起） | ✅ | ✅ | **baostock** |
| A股分钟K线 | 5/15/30/60分钟 | 1/5/15/30/60分钟 | **互补** |
| A股估值指标 | ✅ 日线级 | ✅ | **baostock** |
| A股季频财务（2007~） | ✅ 6大类 | ⚠️ 字段不统一 | **baostock** |
| 复权因子 | ✅ | ❌ | **baostock** |
| 业绩快报/预告 | ✅ | ❌ | **baostock** |
| 行业分类 | ✅ 证监会标准 | ⚠️ 多标准 | **baostock** |
| 港股/美股/基金/期货 | ❌ | ✅ | **akshare** |
| 实时行情 | ❌ | ✅ | **akshare** |
| 资金流向/龙虎榜/融资融券 | ❌ | ✅ | **akshare** |
| 新闻/ESG/另类数据 | ❌ | ✅ | **akshare** |

---

## 已知限制

- RabbitMQ 各 sink 直接发布业务对象，未统一包装 `meta/payload`；下游作业做了兼容处理。
- flink-connector-rabbitmq 3.0.1-1.17 在作业内 failover 后存在 ack 状态错位导致消费静默停滞的缺陷（应急放行 `scripts/flink_queue_drain.sh`，根治需自研 at-least-once Source，详见 `docs/tech_references.md`）。
- `QuestDBSink`/`RabbitMQSink` 仍基于已废弃的 `RichSinkFunction`，Sink V2 迁移待做。

## 后续扩展

- [x] **Phase 8.x**（已完成）：Flink → QuestDB 真实写入、日线 fan-out 派生（周月线重采样/涨跌停价）、CEP 形态检测、DEDUP UPSERT KEYS、新鲜度巡检与缺口自愈
- [ ] **衍生链根治**：KlineResample 查库微批化 + 自研 at-least-once RMQ Source
- [ ] **Phase 9**: K线/因子/文本向量编码 + HNSW 相似性搜索（基于 PostgreSQL + pgvector）
- [ ] **Phase 10**: 测试覆盖、性能基准
- [ ] **FastAPI 服务层**: 统一 REST API，实现数据库解耦

---

## 环境要求

- Python 3.12+ (conda `x_quant`)
- Java 17 + Maven 3.8+
- Docker & Docker Compose
- QuestDB（当前主库，PostgreSQL wire 协议）
- RabbitMQ 3.12+（含 rabbitmq_prometheus 插件）
- Apache Flink 1.20
- PostgreSQL 16 + pgvector 0.8+（向量扩展预留，当前未在 `docker-compose.yml` 中部署）