From 9e694ee34af90f2ba7206350e3c709ce98f8cf0c Mon Sep 17 00:00:00 2001 From: leefer Date: Thu, 30 Jul 2026 11:42:33 +0800 Subject: [PATCH] rebuild(migration): audit all legacy data --- next/docs/evidence/stage-14/acceptance.md | 15 +- .../stage-14/real-migration-report.json | 202 ++++- next/docs/final/subtraction-audit.md | 9 +- next/tests/test_legacy_migration.py | 215 ++++- next/tools/legacy_migration.py | 743 +++++++++++++++++- 5 files changed, 1120 insertions(+), 64 deletions(-) diff --git a/next/docs/evidence/stage-14/acceptance.md b/next/docs/evidence/stage-14/acceptance.md index 21d5dbd..c7dc6c8 100644 --- a/next/docs/evidence/stage-14/acceptance.md +++ b/next/docs/evidence/stage-14/acceptance.md @@ -2,18 +2,20 @@ ## 交付边界 -- 旧系统只作为只读来源；真实演练源为 `data/review.db`，目标为新建的 `next/data/stage14-migrated.db`。 +- 旧系统只作为只读来源；真实演练源为 `data/review.db`，目标为系统临时目录中的全新数据库。 - 迁移逻辑仅存在于一次性工具 `tools/legacy_migration.py`，正常应用不导入旧系统代码或旧数据库。 - 新 Docker、备份和恢复资产仅位于 `next/`，没有停止、修改或替换 NAS 正式容器。 ## 真实迁移结果 -- 源库 SHA-256：`b7c444359c3b9b201a47ce6b23e55ad7ca4eba3d9dd67dfc8544b575d43f4924`。 +- 源库 SHA-256：`5dda9e1538acb7efbcc6a7dbdf201978699daf03ff4ccef4d0677732f2e2f1ab`。正式切换时必须先停写并重新生成最终哈希。 - SQLite `integrity_check=ok`，外键违规 0。 - 3 个账号与会员状态、1 份出生资料、2 项系统数据凭据、3 个模型配置均已转换。 - 6 条自选、3 条复盘、2 条提醒、28 条问师消息、45 条问师偏好、31 条问天历史均已迁移并保留账号归属。 - 196 次历史选股与 16 条手动策略跟踪已迁移；每个历史交易日使用明确标注的归档因子快照，不伪造旧因子覆盖率。 -- 5,534 个股票目录、261 个交易日、24 份市场摘要、5,567 份最近 90 根日 K 展示归档和 24 份兼容市场洞察已迁移。 +- 5,534 个股票目录、394 个题材目录、261 个交易日、25 份市场摘要、5,579 份股票/题材日 K 展示归档和 47 份兼容市场洞察已迁移。 +- 286 条 iFinD 事件原因修订、9 份板块成分股快照、12 份题材详情及成分股、14 个龙虎榜历史日期和 110 份游资档案可被新业务服务直接读取。 +- 真实源库 36 张业务表及 24 种快照全部被分类；未分类表和未分类快照均为 0。报告同时记录逐表源行数、完整目标表行数和每项主动舍弃原因。 - 真实账号密码验证、管理员与永久会员状态、模型选择、问天历史和选股历史抽验通过。 - 逐项机器报告见 `real-migration-report.json`，报告不含明文令牌或模型密钥。 @@ -23,6 +25,7 @@ - 已取消的用户自主 LLM 配置不迁移。 - 旧原始因子表不复制成兼容表；由新系统受治理的数据同步和因子任务重建。 - 旧指数行仅有收盘价，缺少新图表契约要求的 OHLC，未伪造成指数 K 线。 +- 旧库 36 条无用户归属的内置策略不复制；它们由当前唯一版本化产品策略目录取代。用户自建策略才进入 `custom_screener_strategies`，196 次历史运行结果仍完整保留。 - 日 K 展示归档每个标的保留最近 90 根；旧全量数据库和一致性备份继续作为审计资产保留。 ## 备份恢复演练 @@ -43,9 +46,9 @@ ## 质量门 - Ruff：通过。 -- pytest：96 项通过。 +- pytest：106 项通过。 - Vue 类型检查：通过。 - Vitest：3 个文件、7 项通过。 -- Vite 生产构建：通过，CSS 89.25KB，JS 279.23KB，均为构建前原始体积。 -- Playwright：17 项通过，单 worker，耗时 53.6 秒。 +- Vite 生产构建：通过，CSS 92.40KB，JS 299.80KB，均为构建前原始体积。 +- Playwright：26 项通过，单 worker，耗时 147.3 秒。 - `git diff --check` 与已知令牌/密码扫描：通过。 diff --git a/next/docs/evidence/stage-14/real-migration-report.json b/next/docs/evidence/stage-14/real-migration-report.json index cc08e3b..d81dd3c 100644 --- a/next/docs/evidence/stage-14/real-migration-report.json +++ b/next/docs/evidence/stage-14/real-migration-report.json @@ -1,51 +1,209 @@ { "source": "C:\\Users\\MoBai\\Documents\\gupiaofupan\\webapp\\data\\review.db", - "target": "C:\\Users\\MoBai\\Documents\\gupiaofupan\\webapp\\next\\data\\stage14-migrated.db", - "source_sha256": "b7c444359c3b9b201a47ce6b23e55ad7ca4eba3d9dd67dfc8544b575d43f4924", - "target_sha256": "9b0ca7faed7a15b4099bd09ae7a0c94b52a0c8a7dbfedb9d2b26bdf47e34d5d2", + "target": "C:\\Users\\MoBai\\AppData\\Local\\Temp\\xiaobai-next-migration-evidence-aa6d37d29f7a4a77994b04de8b3b3e22.db", + "source_sha256": "5dda9e1538acb7efbcc6a7dbdf201978699daf03ff4ccef4d0677732f2e2f1ab", + "target_sha256": "615992f425bbbc304e9fea8e7d34c42ab8ef7d99156757864bbbc50c80886987", "integrity": "ok", "foreign_key_violations": 0, "migrated": { "alerts": 2, "birth_profiles": 1, - "chart_series": 5567, + "chart_series": 5579, "heaven_readings": 31, "llm_models": 3, "llm_usage_daily": 9, "market_entities": 5534, - "market_insight_snapshots": 24, - "market_summaries": 24, + "market_event_revisions": 286, + "market_insight_snapshots": 47, + "market_summaries": 25, "memberships": 3, "mentor_messages": 28, "mentor_preferences": 45, "review_notes": 3, "screener_runs": 196, + "sector_member_snapshots": 9, "strategy_tracks": 16, "system_credentials": 2, "trading_days": 261, "users": 3, "watchlist_entries": 6 }, - "target_counts": { - "users": 3, - "memberships": 3, - "market_entities": 5537, - "market_summaries": 24, - "chart_series": 5567, - "watchlist_entries": 6, - "review_notes": 3, - "trade_entries": 0, + "source_tables": { "alerts": 2, - "mentor_messages": 28, + "assistant_messages": 0, + "auction_factors": 505070, + "benchmark_bars": 260, + "daily_bars": 1424973, + "daily_indicators": 764710, + "dashboard_snapshots": 25, + "data_snapshots": 422, + "earnings_events": 580, + "fundamental_indicators": 49481, "heaven_readings": 31, + "job_runs": 1101, + "lhb_institution_daily": 47, + "llm_usage": 72, + "mentor_messages": 28, + "mentor_preferences": 45, + "moneyflow_daily": 31175, + "popularity_factors": 232, + "reason_overrides": 0, + "review_notes": 3, + "schema_migrations": 3, "screener_runs": 196, - "custom_screener_strategies": 0, - "strategy_tracks": 16 + "screener_strategies": 36, + "seat_aliases": 0, + "sector_phase_overrides": 0, + "stock_master": 5534, + "strategy_tracks": 16, + "sync_runs": 21053, + "system_settings": 1, + "trade_entries": 0, + "user_birth_profiles": 1, + "user_credentials": 1, + "user_sessions": 207, + "users": 3, + "watchlist": 6, + "wencai_saved_queries": 0 }, - "intentionally_skipped": { - "sessions": "sessions are intentionally invalidated during cutover", + "handled_tables": [ + "alerts", + "assistant_messages", + "daily_bars", + "dashboard_snapshots", + "data_snapshots", + "heaven_readings", + "llm_usage", + "mentor_messages", + "mentor_preferences", + "reason_overrides", + "review_notes", + "screener_runs", + "screener_strategies", + "seat_aliases", + "stock_master", + "strategy_tracks", + "system_settings", + "trade_entries", + "user_birth_profiles", + "users", + "watchlist" + ], + "intentionally_skipped_tables": { + "auction_factors": "governed provider inputs are rebuilt by scheduled jobs", + "benchmark_bars": "legacy rows lack OHLC values required by the chart contract", + "daily_indicators": "governed provider inputs are rebuilt by scheduled jobs", + "earnings_events": "governed provider inputs are rebuilt by scheduled jobs", + "fundamental_indicators": "governed provider inputs are rebuilt by scheduled jobs", + "job_runs": "legacy operational logs are not user-facing history", + "lhb_institution_daily": "governed provider inputs are rebuilt by scheduled jobs", + "moneyflow_daily": "governed provider inputs are rebuilt by scheduled jobs", + "popularity_factors": "governed provider inputs are rebuilt by scheduled jobs", + "schema_migrations": "legacy implementation metadata does not apply to the new schema", + "sector_phase_overrides": "sector phase overrides are no longer product-configurable", + "sync_runs": "legacy operational logs are not user-facing history", "user_credentials": "per-user LLM configuration was removed from the product", - "raw_factor_tables": "reproducible provider inputs are rebuilt by governed sync jobs", - "benchmark_bars": "legacy rows lack OHLC values required by the chart contract" + "user_sessions": "sessions are intentionally invalidated during cutover", + "wencai_saved_queries": "the WenCai feature was explicitly removed from the product" + }, + "unmapped_tables": [], + "source_snapshot_kinds": { + "auction_center_v1": 2, + "auction_center_v2": 1, + "auction_center_v3": 1, + "auction_center_v4": 1, + "auction_center_v5": 5, + "auction_center_v6": 3, + "dashboard_request_v1": 2, + "dragon_tiger": 7, + "heaven_indices": 14, + "heaven_sector": 65, + "heaven_stock": 2, + "hot_money_detail_v2": 8, + "hot_money_detail_v3": 7, + "hot_money_profiles_v1": 1, + "ifind_event_enrichment_v1": 4, + "popularity_v1": 7, + "rotation_sector_members_v1": 9, + "screener_auto_v1": 3, + "search_directory": 1, + "stock_detail": 256, + "stock_intraday": 3, + "theme_detail_v1": 12, + "theme_directory_v1": 1, + "theme_library_v1": 7 + }, + "handled_snapshot_kinds": [ + "auction_center_v1", + "auction_center_v2", + "auction_center_v3", + "auction_center_v4", + "auction_center_v5", + "auction_center_v6", + "dragon_tiger", + "hot_money_detail_v2", + "hot_money_detail_v3", + "hot_money_profiles_v1", + "ifind_event_enrichment_v1", + "popularity_v1", + "rotation_sector_members_v1", + "theme_detail_v1", + "theme_directory_v1", + "theme_library_v1" + ], + "intentionally_skipped_snapshot_kinds": { + "dashboard_request_v1": "request cache is superseded by archived dashboard snapshots", + "heaven_indices": "rebuildable input cache; saved Heaven readings are migrated", + "heaven_sector": "rebuildable input cache; saved Heaven readings are migrated", + "heaven_stock": "rebuildable input cache; saved Heaven readings are migrated", + "screener_auto_v1": "derived cache is superseded by migrated screener runs", + "search_directory": "search data is rebuilt from the migrated entity directory", + "stock_detail": "rebuildable display cache", + "stock_intraday": "rebuildable realtime display cache" + }, + "unmapped_snapshot_kinds": [], + "intentionally_skipped_rows": { + "screener_strategies_builtin": { + "count": 36, + "reason": "legacy built-ins are superseded by the single versioned product catalog; only user-created strategies are migrated" + } + }, + "target_counts": { + "alerts": 2, + "birth_profiles": 1, + "chart_series": 5579, + "custom_screener_strategies": 0, + "heaven_readings": 31, + "job_runs": 0, + "llm_attempts": 0, + "llm_configuration": 1, + "llm_models": 3, + "llm_requests": 0, + "llm_usage_daily": 9, + "market_entities": 5931, + "market_event_revisions": 286, + "market_insight_snapshots": 47, + "market_summaries": 25, + "memberships": 3, + "mentor_messages": 28, + "mentor_preferences": 45, + "review_assistant_messages": 0, + "review_notes": 3, + "schema_migrations": 12, + "screener_factor_snapshots": 10, + "screener_factor_values": 0, + "screener_run_backtests": 0, + "screener_runs": 196, + "seat_aliases": 0, + "sector_member_snapshots": 9, + "sessions": 0, + "strategy_track_bars": 0, + "strategy_track_events": 0, + "strategy_tracks": 16, + "system_credentials": 2, + "trade_entries": 0, + "trading_days": 261, + "users": 3, + "watchlist_entries": 6 } } diff --git a/next/docs/final/subtraction-audit.md b/next/docs/final/subtraction-audit.md index ccb0e20..82ec667 100644 --- a/next/docs/final/subtraction-audit.md +++ b/next/docs/final/subtraction-audit.md @@ -2,8 +2,8 @@ ## 结论 -重建版运行时代码共206个源码文件、25,151行；旧版运行时代码共55个文件、61,793行。 -在功能完整迁移并增加移动端后，新运行时代码减少36,642行，约59.3%。文件数量增加来自按职责拆分， +重建版运行时代码共219个源码文件、25,077行；旧版运行时代码共55个文件、61,793行。 +在功能完整迁移并增加移动端后，新运行时代码减少36,716行，约59.4%。文件数量增加来自按职责拆分， 不再由`server.py`、`app.js`、`styles.css`和覆盖样式承载整个系统。 一次性迁移、备份、测试与文档不计入运行时代码比较；旧系统仍保留作回退资产，但新系统没有运行时导入。 @@ -18,7 +18,7 @@ | 弹窗 | `DialogHost.vue`+UI Store | 无页面级第二套全局弹窗 | | 设计令牌 | `frontend/src/shared/styles/tokens.css` | 主题颜色、字号、间距集中 | | 移动规则 | `frontend/src/shared/styles/mobile.css` | 不复制移动页面或API | -| 数据演进 | `backend/database/migrations/` | 10个有序、带签名migration | +| 数据演进 | `backend/database/migrations/` | 12个有序、带签名migration | 阶段15发现并删除了LLM网关对账户领域具体类的3个反向导入，改为网关内部最小Protocol； 组合根仍注入原服务，没有第二套权限或模型逻辑。基础数据层、数据库层没有反向依赖业务Feature。 @@ -39,7 +39,7 @@ | `screener/technical.py` | 434 | 纯技术因子计算 | 新因子族能形成独立输入契约 | | `mobile.css` | 537 | 唯一跨页移动规则，按断点组织 | 出现页面冲突或超过650行 | | 4个领域CSS | 403-415 | 单一领域且仅略超阈值 | 新增覆盖层或跨领域选择器 | -| `tools/legacy_migration.py` | 629 | 一次性离线适配，不进入运行时 | 新增第二旧版本或需要常驻运行 | +| `tools/legacy_migration.py` | 1,310 | 一次性离线适配；完整覆盖36张表和24种快照，不进入运行时 | 新增第二旧版本或需要常驻运行 | 当前强拆这些文件只会增加接口、跳转和空抽象，不能减少业务复杂度，因此未为满足行数制造目录。 @@ -48,6 +48,7 @@ - 未复制旧`server.py`、`database.py`、`app.js`、15,465行旧样式或8,570行覆盖样式。 - 未保留用户自主LLM配置、模拟行情、公开网页参与正式计算、策略自动跟踪和手动执行盘后策略。 - 未复制旧原始因子表到新运行时；正式因子由唯一数据网关和盘后任务重建。 +- 旧库36条无用户归属的内置策略由当前唯一版本化策略目录取代；历史运行结果保留，不维持第二套策略定义。 - 未引入ORM、通用CRUD、缓存框架、消息队列或第二套前端状态库。 - iFinD、Tushare、东方财富与腾讯职责固定；未做无来源标识的静默多源兜底。 diff --git a/next/tests/test_legacy_migration.py b/next/tests/test_legacy_migration.py index e0189d0..c562e53 100644 --- a/next/tests/test_legacy_migration.py +++ b/next/tests/test_legacy_migration.py @@ -9,6 +9,7 @@ import sqlite3 from cryptography.fernet import Fernet +from backend.features.market.events import apply_event_revisions from backend.security.passwords import PasswordHasher from tools.legacy_migration import LegacyMigrator @@ -81,6 +82,9 @@ def _legacy_database(path, key: str) -> None: entry_price REAL,created_at TEXT,updated_at TEXT); CREATE TABLE data_snapshots (kind TEXT,cache_key TEXT,source TEXT,payload TEXT, updated_at TEXT,PRIMARY KEY(kind,cache_key)); + CREATE TABLE reason_overrides (trade_date TEXT,code TEXT,reason TEXT, + updated_at TEXT,PRIMARY KEY(trade_date,code)); + CREATE TABLE seat_aliases (seat_name TEXT PRIMARY KEY,alias TEXT,updated_at TEXT); """ ) now = "2026-07-29T16:00:00+08:00" @@ -103,8 +107,23 @@ def _legacy_database(path, key: str) -> None: connection.execute( "INSERT INTO daily_bars VALUES ('20260729','000001.SZ',10,11,9,10.5,5,100,1000)" ) + dashboard = json.dumps( + { + "limits": [ + { + "ts_code": "000001.SZ", + "code": "000001", + "name": "Ping An Bank", + "reason": "legacy", + } + ], + "broken": [], + "down_limits": [], + } + ) connection.execute( - "INSERT INTO dashboard_snapshots VALUES ('20260729','tushare','{}',1,?)", (now,) + "INSERT INTO dashboard_snapshots VALUES ('20260729','tushare',?,1,?)", + (dashboard, now), ) connection.execute( "INSERT INTO watchlist VALUES (8,'000001','平安银行','银行','red',?,?, '长期')", @@ -139,6 +158,10 @@ def _legacy_database(path, key: str) -> None: "INSERT INTO screener_strategies VALUES (1,'自定义','','[]','{}',0,?,?,8)", (now, now), ) + connection.execute( + "INSERT INTO screener_strategies VALUES (2,'legacy builtin','','[]','{}',1,?,?,NULL)", + (now, now), + ) connection.execute( "INSERT INTO strategy_tracks VALUES (1,8,1,'20260729','策略','000001.SZ','000001','平安银行','银行',10.5,?,?)", (now, now), @@ -146,6 +169,159 @@ def _legacy_database(path, key: str) -> None: connection.execute( "INSERT INTO data_snapshots VALUES ('popularity_v1','20260729','legacy','{}',?)", (now,) ) + connection.execute( + "INSERT INTO reason_overrides VALUES ('20260729','000001','admin reason',?)", + (now,), + ) + connection.execute( + "INSERT INTO seat_aliases VALUES ('seat-a','trader-a',?)", (now,) + ) + snapshots = { + "ifind_event_enrichment_v1": ( + "20260729", + { + "trade_date": "20260729", + "generated_at": now, + "limits": { + "000001": { + "reason": "ifind reason", + "first_time": "09:31:03", + "last_time": "14:52:01", + "open_times": 1, + } + }, + "broken": {}, + "down_limits": {}, + }, + ), + "rotation_sector_members_v1": ( + "20260729:Bank", + { + "meta": {"sector_code": "801780.SI"}, + "rows": [ + { + "ts_code": "000001.SZ", + "code": "000001", + "name": "Ping An Bank", + "quoted": True, + } + ], + }, + ), + "theme_directory_v1": ( + "ths", + {"items": [{"ts_code": "885001.TI", "name": "Theme A"}]}, + ), + "theme_library_v1": ( + "20260729", + { + "meta": {"updated_at": now}, + "summary": {"theme_count": 1}, + "items": [{"code": "885001.TI", "name": "Theme A"}], + }, + ), + "theme_detail_v1": ( + "20260729:885001.TI", + { + "meta": {"updated_at": now}, + "theme": {"code": "885001.TI", "name": "Theme A"}, + "series": [ + { + "trade_date": "20260728", + "open": 10, + "high": 11, + "low": 9, + "close": 10, + "volume": 100, + }, + { + "trade_date": "20260729", + "open": 10, + "high": 12, + "low": 10, + "close": 11, + "volume": 120, + }, + ], + "members": [ + { + "ts_code": "000001.SZ", + "code": "000001", + "name": "Ping An Bank", + "price": 10.5, + "change": 5, + "amount_billion": 1.2, + "has_quote": True, + } + ], + "summary": {"member_count": 1}, + }, + ), + "hot_money_profiles_v1": ( + "directory", + { + "profiles": [ + { + "name": "trader-a", + "description": "profile", + "organizations": ["seat-a"], + } + ] + }, + ), + "dragon_tiger": ( + "20260729", + { + "meta": {"updated_at": now}, + "rows": [ + { + "ts_code": "000001.SZ", + "name": "Ping An Bank", + "change": 5, + "reason": "listed", + "institutions": [ + { + "seat_name": "seat-a", + "buy_million": 2, + "sell_million": 1, + "net_buy_million": 1, + } + ], + } + ], + }, + ), + "hot_money_detail_v3": ( + "20260729", + { + "meta": {"updated_at": now}, + "traders": [ + { + "name": "trader-a", + "operations": [ + { + "ts_code": "000001.SZ", + "name": "Ping An Bank", + "change": 5, + "seat_name": "seat-a", + "buy_million": 2, + "sell_million": 1, + "net_buy_million": 1, + "reason": "listed", + } + ], + } + ], + }, + ), + } + connection.executemany( + "INSERT INTO data_snapshots VALUES (?,?, 'legacy',?,?)", + [ + (kind, cache_key, json.dumps(payload), now) + for kind, (cache_key, payload) in snapshots.items() + ], + ) def test_legacy_migration_is_idempotent_and_preserves_login(tmp_path) -> None: @@ -158,13 +334,50 @@ def test_legacy_migration_is_idempotent_and_preserves_login(tmp_path) -> None: second = LegacyMigrator(source, target, key).run() assert first["integrity"] == second["integrity"] == "ok" + assert first["unmapped_tables"] == [] + assert first["unmapped_snapshot_kinds"] == [] + assert first["source_tables"]["reason_overrides"] == 1 + assert first["intentionally_skipped_rows"]["screener_strategies_builtin"]["count"] == 1 + assert "market_event_revisions" in first["target_counts"] rendered = json.dumps(first) assert "secret-tushare" not in rendered assert "secret-model" not in rendered with sqlite3.connect(target) as connection: + connection.row_factory = sqlite3.Row password = connection.execute("SELECT password_hash FROM users WHERE id=8").fetchone()[0] assert PasswordHasher().verify("Password123", password) assert connection.execute("SELECT count(*) FROM watchlist_entries").fetchone()[0] == 1 assert connection.execute("SELECT count(*) FROM screener_runs").fetchone()[0] == 1 assert connection.execute("SELECT count(*) FROM strategy_tracks").fetchone()[0] == 1 assert connection.execute("SELECT daily_llm_limit FROM memberships").fetchone()[0] == 61 + assert connection.execute("SELECT count(*) FROM seat_aliases").fetchone()[0] == 1 + assert connection.execute( + "SELECT count(*) FROM sector_member_snapshots" + ).fetchone()[0] == 1 + assert connection.execute( + "SELECT count(*) FROM chart_series WHERE entity_type='theme'" + ).fetchone()[0] == 1 + summary = json.loads( + connection.execute( + "SELECT payload_json FROM market_summaries WHERE trade_date='2026-07-29'" + ).fetchone()[0] + ) + assert summary["limits"][0]["identifier"] == "000001.SZ" + all_revisions = tuple( + connection.execute( + """SELECT * FROM market_event_revisions WHERE trade_date='2026-07-29' + ORDER BY priority DESC,id DESC""" + ).fetchall() + ) + assert len(all_revisions) == 2 + revisions = all_revisions[:1] + revised = apply_event_revisions(summary, revisions) + assert revised["limits"][0]["reason"] == "admin reason" + raw_dragon = json.loads( + connection.execute( + """SELECT payload_json FROM market_insight_snapshots + WHERE kind='dragon-list' AND trade_date='2026-07-29'""" + ).fetchone()[0] + ) + assert raw_dragon["profiles"][0]["desc"] == "profile" + assert raw_dragon["official"][0]["hm_name"] == "trader-a" diff --git a/next/tools/legacy_migration.py b/next/tools/legacy_migration.py index 4222fb3..95d3dd6 100644 --- a/next/tools/legacy_migration.py +++ b/next/tools/legacy_migration.py @@ -17,6 +17,77 @@ from backend.database import MIGRATIONS, Database, MigrationRunner ARCHIVE_VERSION = "legacy-archive-v1" +HANDLED_SOURCE_TABLES = frozenset( + { + "alerts", + "assistant_messages", + "dashboard_snapshots", + "data_snapshots", + "daily_bars", + "heaven_readings", + "llm_usage", + "mentor_messages", + "mentor_preferences", + "reason_overrides", + "review_notes", + "screener_runs", + "screener_strategies", + "seat_aliases", + "stock_master", + "strategy_tracks", + "system_settings", + "trade_entries", + "user_birth_profiles", + "users", + "watchlist", + } +) + +SKIPPED_SOURCE_TABLES = { + "auction_factors": "governed provider inputs are rebuilt by scheduled jobs", + "benchmark_bars": "legacy rows lack OHLC values required by the chart contract", + "daily_indicators": "governed provider inputs are rebuilt by scheduled jobs", + "earnings_events": "governed provider inputs are rebuilt by scheduled jobs", + "fundamental_indicators": "governed provider inputs are rebuilt by scheduled jobs", + "job_runs": "legacy operational logs are not user-facing history", + "lhb_institution_daily": "governed provider inputs are rebuilt by scheduled jobs", + "moneyflow_daily": "governed provider inputs are rebuilt by scheduled jobs", + "popularity_factors": "governed provider inputs are rebuilt by scheduled jobs", + "schema_migrations": "legacy implementation metadata does not apply to the new schema", + "sector_phase_overrides": "sector phase overrides are no longer product-configurable", + "sync_runs": "legacy operational logs are not user-facing history", + "user_credentials": "per-user LLM configuration was removed from the product", + "user_sessions": "sessions are intentionally invalidated during cutover", + "wencai_saved_queries": "the WenCai feature was explicitly removed from the product", +} + +HANDLED_SNAPSHOT_KINDS = frozenset( + { + *(f"auction_center_v{version}" for version in range(1, 7)), + "dragon_tiger", + "hot_money_detail_v2", + "hot_money_detail_v3", + "hot_money_profiles_v1", + "ifind_event_enrichment_v1", + "popularity_v1", + "rotation_sector_members_v1", + "theme_detail_v1", + "theme_directory_v1", + "theme_library_v1", + } +) + +SKIPPED_SNAPSHOT_KINDS = { + "dashboard_request_v1": "request cache is superseded by archived dashboard snapshots", + "heaven_indices": "rebuildable input cache; saved Heaven readings are migrated", + "heaven_sector": "rebuildable input cache; saved Heaven readings are migrated", + "heaven_stock": "rebuildable input cache; saved Heaven readings are migrated", + "screener_auto_v1": "derived cache is superseded by migrated screener runs", + "search_directory": "search data is rebuilt from the migrated entity directory", + "stock_detail": "rebuildable display cache", + "stock_intraday": "rebuildable realtime display cache", +} + def _iso_date(value: Any) -> str: text = str(value or "").strip() @@ -54,20 +125,54 @@ def _table_exists(connection: sqlite3.Connection, table: str) -> bool: ) +def _table_names(connection: sqlite3.Connection) -> tuple[str, ...]: + return tuple( + str(row[0]) + for row in connection.execute( + """SELECT name FROM sqlite_master + WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name""" + ) + ) + + +def _normalize_identifiers(value: Any) -> Any: + if isinstance(value, list): + return [_normalize_identifiers(item) for item in value] + if not isinstance(value, dict): + return value + normalized = {key: _normalize_identifiers(item) for key, item in value.items()} + if normalized.get("ts_code") and not normalized.get("identifier"): + normalized["identifier"] = str(normalized["ts_code"]).upper() + return normalized + + +def _observed_at(payload: dict[str, Any], fallback: str) -> str: + meta = payload.get("meta") if isinstance(payload.get("meta"), dict) else {} + return str( + payload.get("observed_at") + or meta.get("updated_at") + or meta.get("generated_at") + or fallback + ) + + +def _time(value: Any) -> str: + text = str(value or "").strip() + return text[:5] if len(text) >= 5 else "" + + class LegacyMigrator: def __init__(self, source: Path, target: Path, encryption_key: str | None) -> None: self.source_path = source.resolve() self.target_path = target.resolve() self.key = encryption_key self.counts: dict[str, int] = defaultdict(int) - self.skipped: dict[str, str] = { - "sessions": "sessions are intentionally invalidated during cutover", - "user_credentials": "per-user LLM configuration was removed from the product", - "raw_factor_tables": "reproducible provider inputs are rebuilt by governed sync jobs", - "benchmark_bars": "legacy rows lack OHLC values required by the chart contract", - } + self.row_skips: dict[str, dict[str, Any]] = {} self.user_ids: set[int] = set() self.run_ids: set[int] = set() + self.stock_ids: dict[str, str] = {} + self.admin_id: int | None = None + self.dashboard_events: dict[str, dict[str, str]] = {} def run(self) -> dict[str, Any]: if self.source_path == self.target_path: @@ -79,6 +184,38 @@ class LegacyMigrator: source = sqlite3.connect(f"file:{self.source_path.as_posix()}?mode=ro", uri=True) source.row_factory = sqlite3.Row try: + source_tables = _table_names(source) + source_table_counts = { + table: int(source.execute(f"SELECT count(*) FROM {table}").fetchone()[0]) + for table in source_tables + } + snapshot_counts = { + str(row["kind"]): int(row["count"]) + for row in source.execute( + "SELECT kind,count(*) AS count FROM data_snapshots GROUP BY kind ORDER BY kind" + ) + } + handled_tables = sorted(set(source_tables) & HANDLED_SOURCE_TABLES) + skipped_tables = { + table: SKIPPED_SOURCE_TABLES[table] + for table in sorted(set(source_tables) & SKIPPED_SOURCE_TABLES.keys()) + } + unmapped_tables = sorted( + set(source_tables) - HANDLED_SOURCE_TABLES - SKIPPED_SOURCE_TABLES.keys() + ) + handled_snapshots = sorted(set(snapshot_counts) & HANDLED_SNAPSHOT_KINDS) + skipped_snapshots = { + kind: SKIPPED_SNAPSHOT_KINDS[kind] + for kind in sorted(set(snapshot_counts) & SKIPPED_SNAPSHOT_KINDS.keys()) + } + unmapped_snapshots = sorted( + set(snapshot_counts) - HANDLED_SNAPSHOT_KINDS - SKIPPED_SNAPSHOT_KINDS.keys() + ) + if unmapped_tables or unmapped_snapshots: + raise RuntimeError( + "unmapped legacy data: " + f"tables={unmapped_tables}, snapshot_kinds={unmapped_snapshots}" + ) with database.transaction() as target: self._accounts(source, target) self._system_settings(source, target) @@ -91,12 +228,7 @@ class LegacyMigrator: foreign_keys = list(target.execute("PRAGMA foreign_key_check")) target_counts = { table: int(target.execute(f"SELECT count(*) FROM {table}").fetchone()[0]) - for table in ( - "users", "memberships", "market_entities", "market_summaries", - "chart_series", "watchlist_entries", "review_notes", "trade_entries", - "alerts", "mentor_messages", "heaven_readings", "screener_runs", - "custom_screener_strategies", "strategy_tracks", - ) + for table in _table_names(target) } finally: source.close() @@ -110,8 +242,16 @@ class LegacyMigrator: "integrity": integrity, "foreign_key_violations": 0, "migrated": dict(sorted(self.counts.items())), + "source_tables": source_table_counts, + "handled_tables": handled_tables, + "intentionally_skipped_tables": skipped_tables, + "unmapped_tables": unmapped_tables, + "source_snapshot_kinds": snapshot_counts, + "handled_snapshot_kinds": handled_snapshots, + "intentionally_skipped_snapshot_kinds": skipped_snapshots, + "unmapped_snapshot_kinds": unmapped_snapshots, + "intentionally_skipped_rows": self.row_skips, "target_counts": target_counts, - "intentionally_skipped": self.skipped, } def _accounts(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: @@ -149,6 +289,10 @@ class LegacyMigrator: ) self.counts["users"] = len(users) self.counts["memberships"] = len(users) + administrators = [ + int(row["id"]) for row in users if str(row["role"]) == "admin" + ] + self.admin_id = min(administrators or self.user_ids) if _table_exists(source, "user_birth_profiles"): for row in source.execute("SELECT * FROM user_birth_profiles"): target.execute( @@ -245,6 +389,9 @@ class LegacyMigrator: def _market(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: observed = datetime.now().astimezone().isoformat(timespec="seconds") stocks = source.execute("SELECT * FROM stock_master ORDER BY ts_code").fetchall() + self.stock_ids = { + str(row["code"]): str(row["ts_code"]).upper() for row in stocks + } for row in stocks: target.execute( """INSERT INTO market_entities VALUES ('stock',?,?,?,?,?,1,'legacy',?) @@ -279,15 +426,33 @@ class LegacyMigrator: previous = trade_date self.counts["trading_days"] = len(dates) for row in source.execute("SELECT * FROM dashboard_snapshots"): + payload = _normalize_identifiers(_json(row["payload"], {})) + trade_date = _iso_date(row["trade_date"]) + events: dict[str, str] = {} + for event_type, key in ( + ("limit_up", "limits"), + ("broken", "broken"), + ("limit_down", "down_limits"), + ): + for item in payload.get(key) or []: + if not isinstance(item, dict): + continue + identity = str( + item.get("identifier") or item.get("ts_code") or item.get("code") or "" + ).upper() + code = identity.split(".")[0] + if code: + events[code] = event_type + self.dashboard_events[trade_date] = events target.execute( """INSERT INTO market_summaries VALUES (?,?,'archive','legacy',1,?,?) ON CONFLICT(trade_date) DO UPDATE SET observed_at=excluded.observed_at, state='archive',source='legacy',coverage=1,payload_json=excluded.payload_json, created_at=excluded.created_at""", ( - _iso_date(row["trade_date"]), + trade_date, row["updated_at"], - row["payload"], + _dump(payload), row["updated_at"], ), ) @@ -317,6 +482,51 @@ class LegacyMigrator: ) if current: self._save_chart(target, current, points, observed) + self._seat_aliases(source, target) + self._reason_overrides(source, target) + + def _seat_aliases( + self, source: sqlite3.Connection, target: sqlite3.Connection + ) -> None: + if not _table_exists(source, "seat_aliases"): + return + for row in source.execute("SELECT * FROM seat_aliases ORDER BY seat_name"): + target.execute( + """INSERT INTO seat_aliases (seat_name,alias_name,updated_at,updated_by) + VALUES (?,?,?,?) ON CONFLICT(seat_name) DO UPDATE SET + alias_name=excluded.alias_name,updated_at=excluded.updated_at, + updated_by=excluded.updated_by""", + (row["seat_name"], row["alias"], row["updated_at"], self.admin_id), + ) + self.counts["seat_aliases"] += 1 + + def _reason_overrides( + self, source: sqlite3.Connection, target: sqlite3.Connection + ) -> None: + if not _table_exists(source, "reason_overrides"): + return + for row in source.execute("SELECT * FROM reason_overrides ORDER BY trade_date,code"): + trade_date = _iso_date(row["trade_date"]) + code = str(row["code"] or "").split(".")[0] + event_type = self.dashboard_events.get(trade_date, {}).get(code) + if not event_type: + self.counts["reason_overrides_unmatched"] += 1 + continue + inserted = self._save_revision( + target, + trade_date=trade_date, + identifier=self.stock_ids.get(code, str(row["code"]).upper()), + event_type=event_type, + reason=str(row["reason"] or "").strip(), + first_time="", + last_time="", + open_times=None, + source="admin", + priority=100, + created_by=self.admin_id, + created_at=str(row["updated_at"]), + ) + self.counts["market_event_revisions"] += int(inserted) def _save_chart( self, @@ -548,7 +758,17 @@ class LegacyMigrator: ) self.run_ids.add(int(row["id"])) self.counts["screener_runs"] += 1 - for row in source.execute("SELECT * FROM screener_strategies WHERE user_id IS NOT NULL"): + strategy_rows = source.execute("SELECT * FROM screener_strategies").fetchall() + builtins = sum(row["user_id"] is None for row in strategy_rows) + if builtins: + self.row_skips["screener_strategies_builtin"] = { + "count": builtins, + "reason": ( + "legacy built-ins are superseded by the single versioned product catalog; " + "only user-created strategies are migrated" + ), + } + for row in (item for item in strategy_rows if item["user_id"] is not None): target.execute( """INSERT OR REPLACE INTO custom_screener_strategies (id,user_id,name,version,formula_json,created_at,updated_at) @@ -586,24 +806,485 @@ class LegacyMigrator: ) self.counts["strategy_tracks"] += 1 + def _save_revision( + self, + target: sqlite3.Connection, + *, + trade_date: str, + identifier: str, + event_type: str, + reason: str, + first_time: str, + last_time: str, + open_times: int | None, + source: str, + priority: int, + created_by: int | None, + created_at: str, + ) -> bool: + exists = target.execute( + """SELECT 1 FROM market_event_revisions + WHERE trade_date=? AND identifier=? AND event_type=? AND reason=? + AND first_time=? AND last_time=? AND open_times IS ? AND source=? + AND priority=? AND created_by IS ? AND created_at=?""", + ( + trade_date, + identifier, + event_type, + reason, + first_time, + last_time, + open_times, + source, + priority, + created_by, + created_at, + ), + ).fetchone() + if exists: + return False + target.execute( + """INSERT INTO market_event_revisions + (trade_date,identifier,event_type,reason,first_time,last_time,open_times, + source,priority,created_by,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + ( + trade_date, + identifier, + event_type, + reason, + first_time, + last_time, + open_times, + source, + priority, + created_by, + created_at, + ), + ) + return True + def _insights(self, source: sqlite3.Connection, target: sqlite3.Connection) -> None: - mappings = { - "auction_center_v6": "auction", - "theme_library_v1": "themes", - "popularity_v1": "popularity", - "dragon_tiger": "dragon-list", - } - for old_kind, new_kind in mappings.items(): - for row in source.execute("SELECT * FROM data_snapshots WHERE kind=?", (old_kind,)): - trade_date = _iso_date(str(row["cache_key"]).split(":", 1)[0]) - if not trade_date: + self._auction_snapshots(source, target) + self._theme_snapshots(source, target) + self._standard_insight_snapshots(source, target) + self._dragon_snapshots(source, target) + self._sector_member_snapshots(source, target) + self._ifind_event_revisions(source, target) + + def _auction_snapshots( + self, source: sqlite3.Connection, target: sqlite3.Connection + ) -> None: + selected: dict[str, tuple[int, sqlite3.Row]] = {} + for row in source.execute( + "SELECT * FROM data_snapshots WHERE kind LIKE 'auction_center_v%'" + ): + trade_date = _iso_date(str(row["cache_key"]).split(":", 1)[0]) + try: + version = int(str(row["kind"]).rsplit("v", 1)[1]) + except ValueError: + continue + current = selected.get(trade_date) + if trade_date and (current is None or version > current[0]): + selected[trade_date] = (version, row) + for trade_date, (_, row) in sorted(selected.items()): + payload = _normalize_identifiers(_json(row["payload"], {})) + payload.update( + { + "trade_date": trade_date, + "observed_at": _observed_at(payload, str(row["updated_at"])), + "state": "archive", + "message": str(payload.get("message") or ""), + } + ) + payload.setdefault("coverage", 1) + payload.setdefault("dynamic", False) + payload.setdefault("_market_rows", list(payload.get("rows") or [])) + self._save_insight(target, "auction", trade_date, "", payload, 1) + + def _theme_snapshots( + self, source: sqlite3.Connection, target: sqlite3.Connection + ) -> None: + directory_rows = source.execute( + "SELECT * FROM data_snapshots WHERE kind='theme_directory_v1'" + ).fetchall() + for row in directory_rows: + payload = _json(row["payload"], {}) + self._save_theme_entities( + target, list(payload.get("items") or []), str(row["updated_at"]) + ) + for row in source.execute( + "SELECT * FROM data_snapshots WHERE kind='theme_library_v1' ORDER BY cache_key" + ): + trade_date = _iso_date(str(row["cache_key"]).split(":", 1)[0]) + payload = _normalize_identifiers(_json(row["payload"], {})) + payload.update( + { + "trade_date": trade_date, + "observed_at": _observed_at(payload, str(row["updated_at"])), + "state": "archive", + "message": str(payload.get("message") or ""), + } + ) + self._save_theme_entities( + target, list(payload.get("items") or []), str(row["updated_at"]) + ) + self._save_insight(target, "themes", trade_date, "", payload, 1) + for row in source.execute( + "SELECT * FROM data_snapshots WHERE kind='theme_detail_v1' ORDER BY cache_key" + ): + cache_key = str(row["cache_key"]) + raw_date, _, identifier = cache_key.partition(":") + trade_date = _iso_date(raw_date) + legacy = _normalize_identifiers(_json(row["payload"], {})) + theme = dict(legacy.get("theme") or {}) + identifier = (identifier or str(theme.get("code") or "")).upper() + members = [] + for item in legacy.get("members") or []: + if not isinstance(item, dict): continue - target.execute( - """INSERT OR REPLACE INTO market_insight_snapshots - VALUES (?,?,'',?,'archive','legacy',1,?)""", - (new_kind, trade_date, row["updated_at"], row["payload"]), + member_id = str(item.get("identifier") or item.get("ts_code") or "").upper() + members.append( + { + "identifier": member_id, + "code": str(item.get("code") or member_id.split(".")[0]), + "name": str(item.get("name") or ""), + "change": item.get("change"), + "close": item.get("close", item.get("price")), + "amount": ( + float(item["amount_billion"]) * 100_000_000 + if item.get("amount_billion") not in (None, "") + else item.get("amount") + ), + "quoted": bool(item.get("quoted", item.get("has_quote"))), + } ) - self.counts["market_insight_snapshots"] += 1 + summary = dict(legacy.get("summary") or {}) + detail = { + "trade_date": trade_date, + "theme": theme, + "summary": summary, + "members": members, + "message": "" if members else "该题材暂无可核验成分股", + "observed_at": _observed_at(legacy, str(row["updated_at"])), + "state": "archive", + } + self._save_insight(target, "themes", trade_date, identifier, detail, 1) + self._save_theme_chart( + target, + identifier, + list(legacy.get("series") or []), + detail["observed_at"], + ) + + def _save_theme_entities( + self, target: sqlite3.Connection, rows: list[dict[str, Any]], observed_at: str + ) -> None: + for item in rows: + identifier = str(item.get("code") or item.get("ts_code") or "").upper() + name = str(item.get("name") or "").strip() + if not identifier or not name: + continue + target.execute( + """INSERT INTO market_entities + (entity_type,identifier,code,name,search_key,sector,active,source,observed_at) + VALUES ('theme',?,?,?,?,NULL,1,'legacy',?) + ON CONFLICT(entity_type,identifier) DO UPDATE SET + code=excluded.code,name=excluded.name,search_key=excluded.search_key, + active=1,source='legacy',observed_at=excluded.observed_at""", + ( + identifier, + identifier.split(".")[0], + name, + f"{identifier} {name}".casefold(), + observed_at, + ), + ) + + def _save_theme_chart( + self, + target: sqlite3.Connection, + identifier: str, + rows: list[dict[str, Any]], + observed_at: str, + ) -> None: + points = [ + { + "time": _iso_date(item.get("trade_date")), + "open": item.get("open"), + "high": item.get("high"), + "low": item.get("low"), + "close": item.get("close"), + "volume": item.get("volume"), + "amount": item.get("amount"), + "average": None, + } + for item in rows + if _iso_date(item.get("trade_date")) and item.get("close") is not None + ] + if not identifier or not points: + return + target.execute( + """INSERT INTO chart_series + (entity_type,identifier,interval,trade_date,observed_at,source,usage, + adjustment,coverage,payload_json,created_at) + VALUES ('theme',?,'day',?,?,'legacy','display','none',1,?,?) + ON CONFLICT(entity_type,identifier,interval,trade_date) DO UPDATE SET + observed_at=excluded.observed_at,payload_json=excluded.payload_json""", + ( + identifier, + points[-1]["time"], + observed_at, + _dump( + { + "previous_close": points[-2]["close"] if len(points) > 1 else None, + "points": points, + } + ), + observed_at, + ), + ) + self.counts["chart_series"] += 1 + + def _standard_insight_snapshots( + self, source: sqlite3.Connection, target: sqlite3.Connection + ) -> None: + for row in source.execute( + "SELECT * FROM data_snapshots WHERE kind='popularity_v1' ORDER BY cache_key" + ): + trade_date = _iso_date(str(row["cache_key"]).split(":", 1)[0]) + payload = _normalize_identifiers(_json(row["payload"], {})) + payload.update( + { + "trade_date": trade_date, + "observed_at": _observed_at(payload, str(row["updated_at"])), + "state": "archive", + "message": str(payload.get("message") or ""), + } + ) + self._save_insight(target, "popularity", trade_date, "", payload, 1) + + def _dragon_snapshots( + self, source: sqlite3.Connection, target: sqlite3.Connection + ) -> None: + profile_row = source.execute( + """SELECT * FROM data_snapshots WHERE kind='hot_money_profiles_v1' + ORDER BY updated_at DESC LIMIT 1""" + ).fetchone() + profiles = [] + if profile_row: + for item in _json(profile_row["payload"], {}).get("profiles") or []: + profiles.append( + { + "name": str(item.get("name") or ""), + "desc": str(item.get("description") or item.get("desc") or ""), + "orgs": _dump(item.get("organizations") or item.get("orgs") or []), + } + ) + stocks_by_date: dict[str, list[dict[str, Any]]] = {} + seats_by_date: dict[str, list[dict[str, Any]]] = {} + observed_by_date: dict[str, str] = {} + for row in source.execute( + "SELECT * FROM data_snapshots WHERE kind='dragon_tiger' ORDER BY cache_key" + ): + trade_date = _iso_date(str(row["cache_key"]).split(":", 1)[0]) + payload = _json(row["payload"], {}) + observed_by_date[trade_date] = _observed_at(payload, str(row["updated_at"])) + stocks = [] + seats = [] + for item in payload.get("rows") or []: + identifier = str(item.get("ts_code") or item.get("identifier") or "").upper() + stocks.append( + { + "ts_code": identifier, + "name": str(item.get("name") or ""), + "pct_change": item.get("change"), + "reason": str(item.get("reason") or ""), + } + ) + for seat in item.get("institutions") or []: + seats.append( + { + "ts_code": identifier, + "exalter": str(seat.get("seat_name") or ""), + "buy": float(seat.get("buy_million") or 0) * 1_000_000, + "sell": float(seat.get("sell_million") or 0) * 1_000_000, + "net_buy": float(seat.get("net_buy_million") or 0) * 1_000_000, + "reason": str(item.get("reason") or ""), + } + ) + stocks_by_date[trade_date] = stocks + seats_by_date[trade_date] = seats + details: dict[str, tuple[int, sqlite3.Row]] = {} + for row in source.execute( + """SELECT * FROM data_snapshots + WHERE kind IN ('hot_money_detail_v2','hot_money_detail_v3')""" + ): + trade_date = _iso_date(str(row["cache_key"]).split(":", 1)[0]) + version = int(str(row["kind"]).rsplit("v", 1)[1]) + if trade_date not in details or version > details[trade_date][0]: + details[trade_date] = (version, row) + official_by_date: dict[str, list[dict[str, Any]]] = {} + for trade_date, (_, row) in details.items(): + payload = _json(row["payload"], {}) + observed_by_date.setdefault( + trade_date, _observed_at(payload, str(row["updated_at"])) + ) + official = [] + derived_stocks: dict[str, dict[str, Any]] = {} + for trader in payload.get("traders") or []: + trader_name = str(trader.get("name") or "") + for operation in trader.get("operations") or []: + identifier = str( + operation.get("ts_code") or operation.get("identifier") or "" + ).upper() + official.append( + { + "ts_code": identifier, + "ts_name": str(operation.get("name") or ""), + "hm_name": trader_name, + "hm_orgs": str(operation.get("seat_name") or ""), + "buy_amount": float(operation.get("buy_million") or 0) * 1_000_000, + "sell_amount": float(operation.get("sell_million") or 0) * 1_000_000, + "net_amount": float(operation.get("net_buy_million") or 0) + * 1_000_000, + } + ) + if identifier: + derived_stocks[identifier] = { + "ts_code": identifier, + "name": str(operation.get("name") or ""), + "pct_change": operation.get("change"), + "reason": str(operation.get("reason") or ""), + } + official_by_date[trade_date] = official + stocks_by_date.setdefault(trade_date, list(derived_stocks.values())) + seats_by_date.setdefault(trade_date, []) + all_dates = sorted(set(stocks_by_date) | set(official_by_date)) + for trade_date in all_dates: + raw = { + "trade_date": trade_date, + "observed_at": observed_by_date.get( + trade_date, datetime.now().astimezone().isoformat(timespec="seconds") + ), + "state": "archive", + "official": official_by_date.get(trade_date, []), + "profiles": profiles, + "stocks": stocks_by_date.get(trade_date, []), + "seats": seats_by_date.get(trade_date, []), + } + self._save_insight(target, "dragon-list", trade_date, "", raw, 1) + + def _sector_member_snapshots( + self, source: sqlite3.Connection, target: sqlite3.Connection + ) -> None: + for row in source.execute( + """SELECT * FROM data_snapshots + WHERE kind='rotation_sector_members_v1' ORDER BY cache_key""" + ): + raw_date, _, sector_name = str(row["cache_key"]).partition(":") + trade_date = _iso_date(raw_date) + payload = _normalize_identifiers(_json(row["payload"], {})) + meta = payload.get("meta") if isinstance(payload.get("meta"), dict) else {} + payload.update( + { + "trade_date": trade_date, + "sector_name": sector_name, + "observed_at": _observed_at(payload, str(row["updated_at"])), + "source": "legacy", + "coverage": 1, + } + ) + target.execute( + """INSERT INTO sector_member_snapshots + (trade_date,sector_name,sector_code,observed_at,source,coverage,payload_json) + VALUES (?,?,?,?, 'legacy',1,?) + ON CONFLICT(trade_date,sector_name) DO UPDATE SET + sector_code=excluded.sector_code,observed_at=excluded.observed_at, + source=excluded.source,coverage=excluded.coverage,payload_json=excluded.payload_json""", + ( + trade_date, + sector_name, + str(meta.get("sector_code") or meta.get("representative") or ""), + payload["observed_at"], + _dump(payload), + ), + ) + self.counts["sector_member_snapshots"] += 1 + + def _ifind_event_revisions( + self, source: sqlite3.Connection, target: sqlite3.Connection + ) -> None: + for row in source.execute( + """SELECT * FROM data_snapshots + WHERE kind='ifind_event_enrichment_v1' ORDER BY cache_key""" + ): + payload = _json(row["payload"], {}) + trade_date = _iso_date(payload.get("trade_date") or row["cache_key"]) + created_at = str(payload.get("generated_at") or row["updated_at"]) + for event_type, key in ( + ("limit_up", "limits"), + ("broken", "broken"), + ("limit_down", "down_limits"), + ): + values = payload.get(key) or {} + if not isinstance(values, dict): + continue + for raw_identifier, detail in values.items(): + if not isinstance(detail, dict): + continue + useful = any( + detail.get(field) not in (None, "") + for field in ("reason", "first_time", "last_time", "open_times") + ) + if not useful: + continue + code = str(raw_identifier).split(".")[0] + inserted = self._save_revision( + target, + trade_date=trade_date, + identifier=self.stock_ids.get(code, str(raw_identifier).upper()), + event_type=event_type, + reason=str(detail.get("reason") or "").strip(), + first_time=_time(detail.get("first_time")), + last_time=_time(detail.get("last_time")), + open_times=( + int(detail["open_times"]) + if detail.get("open_times") not in (None, "") + else None + ), + source="ifind", + priority=20, + created_by=None, + created_at=created_at, + ) + self.counts["market_event_revisions"] += int(inserted) + + def _save_insight( + self, + target: sqlite3.Connection, + kind: str, + trade_date: str, + entity_key: str, + payload: dict[str, Any], + coverage: float, + ) -> None: + target.execute( + """INSERT INTO market_insight_snapshots + (kind,trade_date,entity_key,observed_at,state,source,coverage,payload_json) + VALUES (?,?,?,?,'archive','legacy',?,?) + ON CONFLICT(kind,trade_date,entity_key) DO UPDATE SET + observed_at=excluded.observed_at,state=excluded.state,source=excluded.source, + coverage=excluded.coverage,payload_json=excluded.payload_json""", + ( + kind, + trade_date, + entity_key, + _observed_at(payload, datetime.now().astimezone().isoformat(timespec="seconds")), + max(0, min(float(coverage), 1)), + _dump(payload), + ), + ) + self.counts["market_insight_snapshots"] += 1 def build_parser() -> argparse.ArgumentParser: