leefer/xiaobaifupan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rebuild(migration): audit all legacy data

Browse Source
This commit is contained in:
leefer
2026-07-30 11:42:33 +08:00
parent 227028f499
commit 9e694ee34a
5 changed files with 1120 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
next/tools/legacy_migration.py
+712 -31
View File
@@ -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: