以性能优先为核心前提,重构模块边界、收敛重复分叉、下沉公共能力。目标:单一职责、单向依赖、可复用底座,杜绝功能零散分布与重复开发。
本项目「能跑、数据在库」,但低层代码缺陷清理之后暴露出的问题已从「写错代码」转为「写对代码但零散分布、重复开发、关键能力未在生产生效」。方案目标不是重写功能,而是收敛结构——把散落在三个入口、两套后端、三份副本里的能力归位到唯一归属模块。
newsflash_backend.py(Scripts/bin)↔ kuaixun-service/app
newsflash_collect.py(3 源)↔ extend/fetchers(12 源)
快讯功能存在两套后端实现(`newsflash_backend.py` 注释自陈「来自 finance-terminal/kuaixun-service」,是历史分叉)与两套采集器(3 源直抓 vs 12 源管线)。同一域内双实现,修复必须双份、演进必然漂移。
sync_bridge.py ↔ pipeline_runner.py ↔ daily_update.sh
生产实际走 launchd 的 `daily_update.sh` 直写 raw 库、不过门控;`sync_bridge.py` 是 App 手动同步入口;`pipeline_runner.py`(号称替代前两者、带四级门控)从未接入 launchd。三条链路并存,治理能力在唯一带门控的入口上却不在生产路径。
cache / validator / reporter
历史上 `cache.py`/`validator.py`/`reporter.py` 在 core、tushare、akshare 三处各存一份。tushare 侧已收敛(574→66 行),akshare 侧是有差异的分支版本,待甄别后统一。`db.py` 下沉 core 已成正确范式,需推广到全部公共模块。
按业务域划分 7 个功能域。每域明确「做什么」与「不做什么」——边界即契约,越界即为缺陷。
| 功能域 | 核心功能 | 职责(做) | 边界(不做) |
|---|---|---|---|
| 数据采集 L1 Ingestion |
结构化行情采集 tushare_fetcher / akshare_fetcher |
拉取日/周/月线、财务、资金流、股东、分红等 84+ 张表,落 raw 库(tushare_data / akshare_data) | 不做清洗、不做门控、不做对外服务、不做推送 |
| 快讯采集 L1 Ingestion |
12 源快讯抓取 extend/fetchers/* |
抓取 cls/新浪/财联社/华尔街见闻/格隆汇/金十等 12 源,进入去重管线 | 不做行情数据、不直接对外、不负责推送路由 |
| 数据治理 L2 Governance |
Pipeline DAG + 四级门控 pipeline / gate / cleaner |
七步流水线编排;gate1 采集/gate2 清洗/gate3 入库/gate4 后置四级质量门控;清洗、插补、离群检测、双源交叉验证;产出 fin_data_clean | 不做采集、不做对外服务;门控是 raw→clean 的唯一闸口 |
| 数据存储 L0 Infra |
三库分层 + 连接池 db / db_pool |
raw(tushare/akshare)→ clean 三级库;连接池、幂等 upsert、日期/排序规则护栏 | 不含业务规则、不含字段语义,只提供读写基础设施 |
| 数据访问 L3 Access |
SDK 统一查询 findatahub-client |
统一查询/导出入口(client / api / mcp_server),隔离底层库细节,SQL 注入防护 | 不做采集、不做治理;是查询的唯一对外出口 |
| 快讯服务 L4 Service |
去重 + AI 增强 + 推送 kuaixun-service/app |
四层去重、翻译/摘要/标签/评分、多渠道推送(Bark/钉钉/飞书/Telegram/企微)、REST(:8100) + WebSocket | 不做行情、不做治理;是快讯全生命周期的唯一后端 |
| 应用客户端 L5 UI |
macOS 客户端 + Vue 前端 App (SwiftUI) / kuaixun-frontend |
进程编排、同步触发、快讯展示、数据库管理、调度、监控、节能;对外呈现 | 不做采集/清洗逻辑本身,只编排子进程与展示数据 |
| 运维可观测 横切 |
审计/迁移/监控 Scripts/bin + health/retry |
调度覆盖审计、数据迁移、性能快照、健康监控、自动恢复、告警 | 横切能力,不参与业务数据流 |
五层单向依赖架构,上层依赖下层、下层不感知上层;findatahub-core 是横切公共底座,被 L1/L2/L3 共同依赖。
依赖分两种:代码依赖(import 关系,单向向下)与运行时交互(进程/网络/协议)。
findatahub-core 是唯一公共底座。所有被 2 个以上消费者使用的模块必须下沉 core,各 fetcher 只保留源专属适配层。
| 公共能力 | 组件 | 职责 | 当前状态 |
|---|---|---|---|
| 数据库访问 | BaseDatabaseManager · safe_identifier | 连接池、幂等 upsert/replace、日期规范化护栏、排序规则、表结构迁移 | 已下沉 akshare/tushare 薄封装 |
| 数据校验 | DataValidator · ValidationReport | 字段完整性、类型、OHLC 规则校验 | 已下沉 tushare 收敛,akshare 待甄别 |
| 报告生成 | ReportGenerator · SyncReport | 同步结果报告、表级统计 | 已下沉 同上 |
| 缓存 | SyncCache · TradeDateCache | 同步状态缓存、交易日历缓存(减少重复同步与 DB 查询) | 已下沉 |
| 流水线编排 | Pipeline · Step · FetchStepProtocol | DAG 编排、并行调度、上下文传递 | 在 core,未接入生产 |
| 质量门控 | gate1-4 · execute_gate | 采集/清洗/入库/后置四级门控 | 在 core,未接入生产 |
| 清洗特征 | DataCleaner · DataImputer · OutlierDetector | 清洗、缺失插补、离群检测 | 在 core |
| 交叉验证 | CrossValidator | 双源数据一致性校验 | 在 core |
| 重试/健康 | RetryManager · HealthMonitor | 分类重试、限流、健康监控 | 在 core |
| 分析统计 | StatsAnalyzer · CorrelationAnalyzer · PCA | 统计、相关性、降维 | 在 core |
tushare_fetcher/core/db.py = 42 行薄封装
akshare_fetcher/core/db.py = 31 行薄封装
fetcher 侧只做:继承 + 源专属表结构适配。逻辑全部在 core,改一处全仓生效。
性能不是事后调优,而是架构契约。以下要点逐一映射到层与组件,越界即回归失败。
去重写入、批量同步全部复用一个连接,禁止每条新建 4~5 次 TCP 握手(历史 500 条≈2000 次握手)。连接池统一超时 + rollback 归还。
增量路径强制 upsert,replace 仅限全量快照表。杜绝「增量 DELETE 全表」的 I/O 风暴与数据损毁。
Pipeline 整批写、单事务提交,避免逐行 commit;并行步骤产物按步骤名隔离,防串源重算。
进程健康用心跳文件判定,不用「最后写库时间」(历史 300s 阈值误判 193 次/日)。交易/非交易时段差异化节能退避(30s→3600s)。
TradeDateCache 缓存交易日历、SyncCache 缓存同步状态,避免重复拉取与重复同步;MySQL 缓冲池 2GB 命中优先。
搜索保留 LIKE 连续子串语义(正确优先),拒绝 ngram 全文索引静默丢结果;total 单独 COUNT,不扫全表。
快讯采集、内容增强、推送走 asyncio 并发;限流 + 分类重试(限频/网络/业务分类退避),避免串行阻塞。
MySQLOnDemandManager / PowerSaver:非交易时段降频、按需拉起 mysqld,主进程 UNIX 调用从 5092/s 降至 0.2/s。
| 性能基线 | 非交易时段 | 交易时段(优化前) | 交易时段(优化后) |
|---|---|---|---|
| 主进程 UNIX 调用/s | 0.2–33 | 5092.5 | 0.2 |
| 采集器 UNIX 调用/s | 0.2 | 658.1 | 1.6 |
| MySQL UNIX 调用/s | 42 | 8932 | 48.2 |
| MySQL RSS | 470 MB | 1870 MB | 2959 MB(缓冲池 2GB + 数据增长,非泄漏) |
数据来源:`Scripts/bin/perf_snapshot.py`(libproc 采样)。交易/非交易时段必须分别验证,不可直接比较——交易时段高负载是常态非泄漏。
| 批次 | 动作 | 内容 | 优先级 |
|---|---|---|---|
| 第 1 批 | 快讯域收敛 | 废弃 Scripts/bin/newsflash_backend.py(历史分叉),统一到 kuaixun-service;废弃 3 源 newsflash_collect.py,统一到 extend/fetchers 12 源管线 | P0 结构 |
| 第 2 批 | 同步域收敛 | pipeline_runner.py 装入 launchd,成为 fin_data_clean 唯一写入方;daily_update.sh 降级为 raw 采集(本就该是 raw);sync_bridge.py 收敛为 pipeline 的 App 触发入口 | P0 结构 |
| 第 3 批 | 公共能力收敛 | 甄别 akshare 侧 cache/validator/reporter 分支版本,合并到 core;治理引擎(pipeline/gate)确认唯一运行时位置 | P1 |
| 第 4 批 | 超大文件拆分 | akshare fetcher.py(1035)、newsflash_collect.py(962)、PythonProcessManager.swift(842)、DatabaseSetupGuideView.swift(807) 按职责拆 extension | P1 |
| 第 5 批 | 契约测试补全 | 为公共组件(db/validator/gate/pipeline)补接口契约测试,锁定「边界即护栏」,越界即失败 | P2 |