"""Incrementally mirror front-adjusted K-line data from the shared source DB. This script is intentionally conservative: - The source MySQL is read-only from this script. It never writes to the colleague-owned database and never calls the HTTP backfill endpoints. - Data is written first to local mirror tables suffixed with ``_front_sync`` plus a sync run log table. When ``--append-to-main`` is passed, only missing primary keys are appended into local main tables; existing local rows are never updated, replaced, or deleted. - The sync is idempotent by primary key and can be scheduled repeatedly. Required passwords are read from environment variables or CLI arguments. The script does not hard-code credentials. """ from __future__ import annotations import argparse import json import os import subprocess import tempfile from dataclasses import dataclass from datetime import date, datetime, time, timedelta from pathlib import Path from typing import Iterable SCRIPT_PATH = Path(__file__).absolute() ROOT = SCRIPT_PATH.parents[2] RESULT_ROOT = ROOT / "data" / "result" / "stat" DEFAULT_MYSQL_EXE = Path(os.environ.get("TIANXIA_MYSQL_EXE", r"M:\mysql\server\bin\mysql.exe")) LOCAL_DAILY_TABLE = "a_share_daily_price_front_sync" LOCAL_MINUTE_TABLE = "a_share_minute_price_front_sync" LOCAL_RUN_TABLE = "a_share_kline_front_sync_run" MAIN_DAILY_TABLE = "a_share_daily_price" MAIN_MINUTE_TABLE = "a_share_minute_price" SOURCE_DAILY_TABLE = "cn_stock_kline_1d_front" SOURCE_MINUTE_TABLE = "cn_stock_kline_1m_front" SOURCE_COVERAGE_TABLE = "cn_stock_kline_front_coverage_daily" @dataclass(frozen=True) class MysqlConn: mysql_exe: Path host: str port: str user: str password: str | None database: str def sql_literal(value: object) -> str: if value is None: return "NULL" text = str(value) return "'" + text.replace("\\", "\\\\").replace("'", "''") + "'" def parse_date(value: str) -> date: return datetime.strptime(value, "%Y-%m-%d").date() def date_to_str(value: date) -> str: return value.strftime("%Y-%m-%d") def mysql_cmd(conn: MysqlConn, extra: Iterable[str] | None = None) -> list[str]: cmd = [ str(conn.mysql_exe), "--protocol=TCP", "--local-infile=1", "--compress", f"--host={conn.host}", f"--port={conn.port}", f"--user={conn.user}", "--default-character-set=utf8mb4", ] if extra: cmd.extend(extra) cmd.append(conn.database) return cmd def mysql_env(conn: MysqlConn) -> dict[str, str]: env = os.environ.copy() if conn.password: env["MYSQL_PWD"] = conn.password return env def run_mysql(conn: MysqlConn, sql: str, *, batch: bool = False, skip_column_names: bool = False) -> str: extra = [] if batch: extra.extend(["--batch", "--raw"]) if skip_column_names: extra.append("--skip-column-names") cmd = mysql_cmd(conn, extra + ["-e", sql]) proc = subprocess.run( cmd, text=True, capture_output=True, env=mysql_env(conn), check=False, ) if proc.returncode != 0: raise RuntimeError(proc.stderr.strip() or proc.stdout.strip()) return proc.stdout def query_rows(conn: MysqlConn, sql: str) -> list[dict[str, str]]: out = run_mysql(conn, sql, batch=True) lines = [line for line in out.splitlines() if line.strip()] if not lines: return [] header = lines[0].split("\t") rows: list[dict[str, str]] = [] for line in lines[1:]: values = line.split("\t") rows.append({header[i]: values[i] if i < len(values) else "" for i in range(len(header))}) return rows def scalar(conn: MysqlConn, sql: str, field: str = "v") -> str: rows = query_rows(conn, sql) return rows[0].get(field, "") if rows else "" def export_tsv(conn: MysqlConn, sql: str, out_path: Path) -> None: cmd = mysql_cmd(conn, ["--batch", "--raw", "--quick", "--skip-column-names", "-e", sql]) with out_path.open("wb") as fh: proc = subprocess.run(cmd, stdout=fh, stderr=subprocess.PIPE, env=mysql_env(conn), check=False) if proc.returncode != 0: stderr = proc.stderr.decode("utf-8", errors="replace") raise RuntimeError(stderr.strip()) def local_exec(conn: MysqlConn, sql: str) -> str: return run_mysql(conn, sql) def create_local_schema(conn: MysqlConn) -> None: local_exec( conn, f""" CREATE TABLE IF NOT EXISTS {LOCAL_DAILY_TABLE} ( trade_date DATE NOT NULL, symbol CHAR(9) NOT NULL, open_price DECIMAL(12,4) NULL, high_price DECIMAL(12,4) NULL, low_price DECIMAL(12,4) NULL, close_price DECIMAL(12,4) NULL, pre_close_price DECIMAL(12,4) NULL, volume BIGINT UNSIGNED NULL, amount DECIMAL(20,2) NULL, turnover DOUBLE NULL, source_batch_id VARCHAR(64) NULL, source_table VARCHAR(64) NOT NULL DEFAULT '{SOURCE_DAILY_TABLE}', source_updated_at DATETIME NULL, sync_run_id VARCHAR(96) NOT NULL, synced_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (trade_date, symbol), KEY idx_symbol_date (symbol, trade_date), KEY idx_sync_run (sync_run_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; CREATE TABLE IF NOT EXISTS {LOCAL_MINUTE_TABLE} ( trade_date DATE NOT NULL, trade_time TIME NOT NULL, bar_time DATETIME NOT NULL, symbol CHAR(9) NOT NULL, open_price DECIMAL(12,4) NULL, high_price DECIMAL(12,4) NULL, low_price DECIMAL(12,4) NULL, close_price DECIMAL(12,4) NULL, volume BIGINT UNSIGNED NULL, amount DECIMAL(20,2) NULL, turnover DOUBLE NULL, source_batch_id VARCHAR(64) NULL, source_table VARCHAR(64) NOT NULL DEFAULT '{SOURCE_MINUTE_TABLE}', source_updated_at DATETIME NULL, sync_run_id VARCHAR(96) NOT NULL, synced_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (trade_date, trade_time, symbol), KEY idx_symbol_bar_time (symbol, bar_time), KEY idx_bar_time (bar_time), KEY idx_sync_run (sync_run_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; CREATE TABLE IF NOT EXISTS {LOCAL_RUN_TABLE} ( run_id VARCHAR(96) NOT NULL, started_at DATETIME NOT NULL, finished_at DATETIME NULL, status VARCHAR(64) NOT NULL, periods VARCHAR(64) NOT NULL, start_date DATE NULL, end_date DATE NULL, symbols_scope TEXT NULL, source_host VARCHAR(255) NOT NULL, source_database VARCHAR(128) NOT NULL, local_database VARCHAR(128) NOT NULL, daily_rows_source BIGINT NOT NULL DEFAULT 0, daily_rows_loaded BIGINT NOT NULL DEFAULT 0, minute_rows_source BIGINT NOT NULL DEFAULT 0, minute_rows_loaded BIGINT NOT NULL DEFAULT 0, summary_json JSON NULL, error_message TEXT NULL, PRIMARY KEY (run_id), KEY idx_started_at (started_at), KEY idx_status (status) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, ) def table_exists(conn: MysqlConn, table_name: str) -> bool: value = scalar( conn, "SELECT COUNT(*) AS v FROM information_schema.tables " f"WHERE table_schema = DATABASE() AND table_name = {sql_literal(table_name)}", ) return value not in ("", "0") def max_date(conn: MysqlConn, table_name: str, column_name: str) -> str: if not table_exists(conn, table_name): return "" return scalar(conn, f"SELECT DATE(MAX({column_name})) AS v FROM {table_name}") def max_successful_all_a_sync_date(conn: MysqlConn, period: str) -> str: if not table_exists(conn, LOCAL_RUN_TABLE): return "" if period == "1d" and table_exists(conn, LOCAL_DAILY_TABLE): return scalar( conn, f""" SELECT DATE(MAX(d.trade_date)) AS v FROM {LOCAL_DAILY_TABLE} d JOIN {LOCAL_RUN_TABLE} r ON r.run_id = d.sync_run_id WHERE r.status = 'SUCCESS' AND r.symbols_scope = 'ALL_A' AND FIND_IN_SET('1d', r.periods) > 0 """, ) if period == "1m" and table_exists(conn, LOCAL_MINUTE_TABLE): return scalar( conn, f""" SELECT DATE(MAX(m.trade_date)) AS v FROM {LOCAL_MINUTE_TABLE} m JOIN {LOCAL_RUN_TABLE} r ON r.run_id = m.sync_run_id WHERE r.status = 'SUCCESS' AND r.symbols_scope = 'ALL_A' AND FIND_IN_SET('1m', r.periods) > 0 """, ) return "" def source_max_date(conn: MysqlConn, period: str) -> str: if period == "1d": return scalar(conn, f"SELECT DATE(MAX(trade_date)) AS v FROM {SOURCE_DAILY_TABLE}") if period == "1m": return scalar(conn, f"SELECT DATE(MAX(bar_time)) AS v FROM {SOURCE_MINUTE_TABLE}") raise ValueError(f"unsupported period: {period}") def resolve_start_date(local_conn: MysqlConn, period: str, explicit_start: str | None) -> date: if explicit_start: return parse_date(explicit_start) if period == "1d": local_latest = max_successful_all_a_sync_date(local_conn, "1d") or max_date(local_conn, "a_share_daily_price", "trade_date") elif period == "1m": local_latest = max_successful_all_a_sync_date(local_conn, "1m") or max_date(local_conn, "a_share_minute_price", "trade_date") else: raise ValueError(f"unsupported period: {period}") if not local_latest: raise RuntimeError(f"cannot infer start date for {period}; pass --start-date explicitly") return parse_date(local_latest) + timedelta(days=1) def resolve_end_date(source_conn: MysqlConn, period: str, explicit_end: str | None) -> date: if explicit_end: return parse_date(explicit_end) latest = source_max_date(source_conn, period) if not latest: raise RuntimeError(f"source has no max date for {period}") return parse_date(latest) def date_batches(start: date, end: date, max_days: int) -> Iterable[tuple[date, date]]: cur = start while cur <= end: batch_end = min(cur + timedelta(days=max_days - 1), end) yield cur, batch_end cur = batch_end + timedelta(days=1) def datetime_batches(start: datetime, end: datetime, max_minutes: int) -> Iterable[tuple[datetime, datetime]]: cur = start delta = timedelta(minutes=max_minutes) while cur < end: batch_end = min(cur + delta, end) yield cur, batch_end cur = batch_end def market_datetime_batches(day: date, max_minutes: int) -> Iterable[tuple[datetime, datetime]]: """Yield only likely A-share minute windows, avoiding overnight empty scans.""" sessions = [ (time(9, 30), time(11, 31)), (time(13, 0), time(15, 1)), ] for session_start, session_end in sessions: yield from datetime_batches(datetime.combine(day, session_start), datetime.combine(day, session_end), max_minutes) def symbol_filter(symbols: list[str]) -> str: if not symbols: return "" values = ",".join(sql_literal(symbol.strip()) for symbol in symbols if symbol.strip()) if not values: return "" return f" AND symbol IN ({values})" def count_source_rows(source_conn: MysqlConn, period: str, start: date, end: date, symbols: list[str]) -> int: filt = symbol_filter(symbols) if period == "1d": sql = ( f"SELECT COUNT(*) AS v FROM {SOURCE_DAILY_TABLE} " f"WHERE trade_date BETWEEN {sql_literal(date_to_str(start))} AND {sql_literal(date_to_str(end))}{filt}" ) else: end_next = end + timedelta(days=1) sql = ( f"SELECT COUNT(*) AS v FROM {SOURCE_MINUTE_TABLE} " f"WHERE bar_time >= {sql_literal(date_to_str(start) + ' 00:00:00')} " f"AND bar_time < {sql_literal(date_to_str(end_next) + ' 00:00:00')}{filt}" ) return int(scalar(source_conn, sql) or "0") def source_minute_coverage_rows(source_conn: MysqlConn, start: date, end: date) -> list[dict[str, str]]: try: return query_rows( source_conn, f""" SELECT trade_date, row_count FROM {SOURCE_COVERAGE_TABLE} WHERE period = '1m' AND trade_date BETWEEN {sql_literal(date_to_str(start))} AND {sql_literal(date_to_str(end))} AND row_count > 0 ORDER BY trade_date """, ) except RuntimeError: return [] def source_minute_trade_dates(source_conn: MysqlConn, start: date, end: date, symbols: list[str]) -> list[date]: if not symbols: rows = source_minute_coverage_rows(source_conn, start, end) dates = [parse_date(row["trade_date"]) for row in rows if row.get("trade_date")] if dates: return dates days: list[date] = [] cur = start while cur <= end: days.append(cur) cur += timedelta(days=1) return days def source_symbols_from_daily(source_conn: MysqlConn, start: date, end: date) -> list[str]: rows = query_rows( source_conn, f""" SELECT DISTINCT symbol FROM {SOURCE_DAILY_TABLE} WHERE trade_date BETWEEN {sql_literal(date_to_str(start))} AND {sql_literal(date_to_str(end))} ORDER BY symbol """, ) return [row["symbol"] for row in rows if row.get("symbol")] def chunked(items: list[str], size: int) -> Iterable[list[str]]: if size <= 0: size = 50 for idx in range(0, len(items), size): yield items[idx : idx + size] def count_source_minute_window(source_conn: MysqlConn, start_dt: datetime, end_dt: datetime, symbols: list[str]) -> int: filt = symbol_filter(symbols) sql = ( f"SELECT COUNT(*) AS v FROM {SOURCE_MINUTE_TABLE} " f"WHERE bar_time >= {sql_literal(start_dt.strftime('%Y-%m-%d %H:%M:%S'))} " f"AND bar_time < {sql_literal(end_dt.strftime('%Y-%m-%d %H:%M:%S'))}{filt}" ) return int(scalar(source_conn, sql) or "0") def count_tsv_rows(path: Path) -> int: with path.open("rb") as fh: return sum(1 for _ in fh) def build_daily_select(start: date, end: date, symbols: list[str]) -> str: filt = symbol_filter(symbols) return f""" SELECT symbol, DATE_FORMAT(trade_date, '%Y-%m-%d'), COALESCE(CAST(`open` AS CHAR), '\\\\N'), COALESCE(CAST(high AS CHAR), '\\\\N'), COALESCE(CAST(low AS CHAR), '\\\\N'), COALESCE(CAST(`close` AS CHAR), '\\\\N'), COALESCE(CAST(pre_close AS CHAR), '\\\\N'), COALESCE(CAST(volume AS CHAR), '\\\\N'), COALESCE(CAST(amount AS CHAR), '\\\\N'), COALESCE(CAST(turnover AS CHAR), '\\\\N'), COALESCE(source_batch_id, '\\\\N'), COALESCE(DATE_FORMAT(updated_at, '%Y-%m-%d %H:%i:%s'), '\\\\N') FROM {SOURCE_DAILY_TABLE} WHERE trade_date BETWEEN {sql_literal(date_to_str(start))} AND {sql_literal(date_to_str(end))}{filt} """.strip() def build_minute_select(start: date, end: date, symbols: list[str]) -> str: filt = symbol_filter(symbols) end_next = end + timedelta(days=1) return f""" SELECT symbol, bar_time, COALESCE(CAST(`open` AS CHAR), '\\\\N'), COALESCE(CAST(high AS CHAR), '\\\\N'), COALESCE(CAST(low AS CHAR), '\\\\N'), COALESCE(CAST(`close` AS CHAR), '\\\\N'), COALESCE(CAST(volume AS CHAR), '\\\\N'), COALESCE(CAST(amount AS CHAR), '\\\\N'), COALESCE(CAST(turnover AS CHAR), '\\\\N'), COALESCE(source_batch_id, '\\\\N'), COALESCE(DATE_FORMAT(updated_at, '%Y-%m-%d %H:%i:%s'), '\\\\N') FROM {SOURCE_MINUTE_TABLE} FORCE INDEX(idx_bar_time_symbol) WHERE bar_time >= {sql_literal(date_to_str(start) + ' 00:00:00')} AND bar_time < {sql_literal(date_to_str(end_next) + ' 00:00:00')}{filt} """.strip() def build_minute_select_window(start_dt: datetime, end_dt: datetime, symbols: list[str]) -> str: filt = symbol_filter(symbols) index_hint = "FORCE INDEX(idx_symbol_bar_time)" if symbols else "FORCE INDEX(idx_bar_time_symbol)" return f""" SELECT symbol, bar_time, COALESCE(CAST(`open` AS CHAR), '\\\\N'), COALESCE(CAST(high AS CHAR), '\\\\N'), COALESCE(CAST(low AS CHAR), '\\\\N'), COALESCE(CAST(`close` AS CHAR), '\\\\N'), COALESCE(CAST(volume AS CHAR), '\\\\N'), COALESCE(CAST(amount AS CHAR), '\\\\N'), COALESCE(CAST(turnover AS CHAR), '\\\\N'), COALESCE(source_batch_id, '\\\\N'), COALESCE(DATE_FORMAT(updated_at, '%Y-%m-%d %H:%i:%s'), '\\\\N') FROM {SOURCE_MINUTE_TABLE} {index_hint} WHERE bar_time >= {sql_literal(start_dt.strftime('%Y-%m-%d %H:%M:%S'))} AND bar_time < {sql_literal(end_dt.strftime('%Y-%m-%d %H:%M:%S'))}{filt} """.strip() def load_daily(local_conn: MysqlConn, tsv_path: Path, run_id: str) -> None: path = str(tsv_path).replace("\\", "/").replace("'", "''") local_exec( local_conn, f""" LOAD DATA LOCAL INFILE '{path}' REPLACE INTO TABLE {LOCAL_DAILY_TABLE} CHARACTER SET utf8mb4 FIELDS TERMINATED BY '\t' ESCAPED BY '\\\\' LINES TERMINATED BY '\n' (symbol, trade_date, open_price, high_price, low_price, close_price, pre_close_price, volume, amount, turnover, source_batch_id, source_updated_at) SET source_table = '{SOURCE_DAILY_TABLE}', sync_run_id = {sql_literal(run_id)}, synced_at = CURRENT_TIMESTAMP; """, ) def load_minute(local_conn: MysqlConn, tsv_path: Path, run_id: str) -> None: path = str(tsv_path).replace("\\", "/").replace("'", "''") local_exec( local_conn, f""" LOAD DATA LOCAL INFILE '{path}' REPLACE INTO TABLE {LOCAL_MINUTE_TABLE} CHARACTER SET utf8mb4 FIELDS TERMINATED BY '\t' ESCAPED BY '\\\\' LINES TERMINATED BY '\n' (symbol, @bar_time, open_price, high_price, low_price, close_price, volume, amount, turnover, source_batch_id, source_updated_at) SET trade_date = DATE(@bar_time), trade_time = TIME(@bar_time), bar_time = @bar_time, source_table = '{SOURCE_MINUTE_TABLE}', sync_run_id = {sql_literal(run_id)}, synced_at = CURRENT_TIMESTAMP; """, ) def local_loaded_count(local_conn: MysqlConn, period: str, start: date, end: date, run_id: str) -> int: if period == "1d": sql = ( f"SELECT COUNT(*) AS v FROM {LOCAL_DAILY_TABLE} " f"WHERE trade_date BETWEEN {sql_literal(date_to_str(start))} AND {sql_literal(date_to_str(end))} " f"AND sync_run_id = {sql_literal(run_id)}" ) else: sql = ( f"SELECT COUNT(*) AS v FROM {LOCAL_MINUTE_TABLE} " f"WHERE trade_date BETWEEN {sql_literal(date_to_str(start))} AND {sql_literal(date_to_str(end))} " f"AND sync_run_id = {sql_literal(run_id)}" ) return int(scalar(local_conn, sql) or "0") def local_loaded_minute_window(local_conn: MysqlConn, start_dt: datetime, end_dt: datetime, run_id: str) -> int: sql = ( f"SELECT COUNT(*) AS v FROM {LOCAL_MINUTE_TABLE} " f"WHERE bar_time >= {sql_literal(start_dt.strftime('%Y-%m-%d %H:%M:%S'))} " f"AND bar_time < {sql_literal(end_dt.strftime('%Y-%m-%d %H:%M:%S'))} " f"AND sync_run_id = {sql_literal(run_id)}" ) return int(scalar(local_conn, sql) or "0") def append_daily_to_main(local_conn: MysqlConn, start: date, end: date, run_id: str) -> dict[str, int]: null_ohlc_rows = int( scalar( local_conn, f""" SELECT COUNT(*) AS v FROM {LOCAL_DAILY_TABLE} WHERE sync_run_id = {sql_literal(run_id)} AND trade_date BETWEEN {sql_literal(date_to_str(start))} AND {sql_literal(date_to_str(end))} AND (open_price IS NULL OR high_price IS NULL OR low_price IS NULL OR close_price IS NULL) """, ) or "0" ) inserted_rows = int( scalar( local_conn, f""" INSERT INTO {MAIN_DAILY_TABLE} (trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount) SELECT s.trade_date, s.symbol, s.open_price, s.high_price, s.low_price, s.close_price, s.volume, s.amount FROM {LOCAL_DAILY_TABLE} s LEFT JOIN {MAIN_DAILY_TABLE} d ON d.trade_date = s.trade_date AND d.symbol = s.symbol WHERE s.sync_run_id = {sql_literal(run_id)} AND s.trade_date BETWEEN {sql_literal(date_to_str(start))} AND {sql_literal(date_to_str(end))} AND d.symbol IS NULL AND s.open_price IS NOT NULL AND s.high_price IS NOT NULL AND s.low_price IS NOT NULL AND s.close_price IS NOT NULL; SELECT ROW_COUNT() AS v; """, ) or "0" ) return {"appended_rows": inserted_rows, "null_ohlc_skipped_rows": null_ohlc_rows} def append_minute_window_to_main( local_conn: MysqlConn, start_dt: datetime, end_dt: datetime, run_id: str, symbols: list[str] | None = None, ) -> dict[str, int]: filter_values = ",".join(sql_literal(symbol.strip()) for symbol in (symbols or []) if symbol.strip()) filt = f"AND symbol IN ({filter_values})" if filter_values else "" source_alias_filt = f"AND s.symbol IN ({filter_values})" if filter_values else "" null_ohlc_rows = int( scalar( local_conn, f""" SELECT COUNT(*) AS v FROM {LOCAL_MINUTE_TABLE} WHERE sync_run_id = {sql_literal(run_id)} AND bar_time >= {sql_literal(start_dt.strftime('%Y-%m-%d %H:%M:%S'))} AND bar_time < {sql_literal(end_dt.strftime('%Y-%m-%d %H:%M:%S'))} {filt} AND (open_price IS NULL OR high_price IS NULL OR low_price IS NULL OR close_price IS NULL) """, ) or "0" ) inserted_rows = int( scalar( local_conn, f""" INSERT INTO {MAIN_MINUTE_TABLE} (trade_date, trade_time, symbol, open_price, high_price, low_price, close_price, volume, amount, open_interest) SELECT s.trade_date, s.trade_time, s.symbol, s.open_price, s.high_price, s.low_price, s.close_price, s.volume, s.amount, NULL FROM {LOCAL_MINUTE_TABLE} s LEFT JOIN {MAIN_MINUTE_TABLE} m ON m.trade_date = s.trade_date AND m.trade_time = s.trade_time AND m.symbol = s.symbol WHERE s.sync_run_id = {sql_literal(run_id)} AND s.bar_time >= {sql_literal(start_dt.strftime('%Y-%m-%d %H:%M:%S'))} AND s.bar_time < {sql_literal(end_dt.strftime('%Y-%m-%d %H:%M:%S'))} {source_alias_filt} AND m.symbol IS NULL AND s.open_price IS NOT NULL AND s.high_price IS NOT NULL AND s.low_price IS NOT NULL AND s.close_price IS NOT NULL; SELECT ROW_COUNT() AS v; """, ) or "0" ) return {"appended_rows": inserted_rows, "null_ohlc_skipped_rows": null_ohlc_rows} def insert_run_log_start( local_conn: MysqlConn, run_id: str, periods: list[str], start: date | None, end: date | None, symbols: list[str], source_conn: MysqlConn, ) -> None: local_exec( local_conn, f""" REPLACE INTO {LOCAL_RUN_TABLE} (run_id, started_at, status, periods, start_date, end_date, symbols_scope, source_host, source_database, local_database) VALUES ( {sql_literal(run_id)}, NOW(), 'RUNNING', {sql_literal(','.join(periods))}, {sql_literal(date_to_str(start)) if start else 'NULL'}, {sql_literal(date_to_str(end)) if end else 'NULL'}, {sql_literal(','.join(symbols) if symbols else 'ALL_A')}, {sql_literal(source_conn.host + ':' + source_conn.port)}, {sql_literal(source_conn.database)}, DATABASE() ); """, ) def update_run_log_finish( local_conn: MysqlConn, run_id: str, status: str, summary: dict[str, object], error_message: str | None = None, ) -> None: daily = summary.get("periods", {}).get("1d", {}) if isinstance(summary.get("periods"), dict) else {} minute = summary.get("periods", {}).get("1m", {}) if isinstance(summary.get("periods"), dict) else {} local_exec( local_conn, f""" UPDATE {LOCAL_RUN_TABLE} SET finished_at = NOW(), status = {sql_literal(status)}, daily_rows_source = {int(daily.get('source_rows', 0) or 0)}, daily_rows_loaded = {int(daily.get('loaded_rows', 0) or 0)}, minute_rows_source = {int(minute.get('source_rows', 0) or 0)}, minute_rows_loaded = {int(minute.get('loaded_rows', 0) or 0)}, summary_json = CAST({sql_literal(json.dumps(summary, ensure_ascii=False))} AS JSON), error_message = {sql_literal(error_message)} WHERE run_id = {sql_literal(run_id)}; """, ) def sync_period( source_conn: MysqlConn, local_conn: MysqlConn, period: str, start: date, end: date, symbols: list[str], run_id: str, max_days: int, minute_max_minutes: int, minute_symbols_per_batch: int, dry_run: bool, append_to_main: bool, ) -> dict[str, object]: if period == "1m": coverage_rows = source_minute_coverage_rows(source_conn, start, end) if not symbols else [] source_rows_total = sum(int(row.get("row_count") or 0) for row in coverage_rows) if coverage_rows else 0 else: coverage_rows = [] source_rows_total = count_source_rows(source_conn, period, start, end, symbols) result: dict[str, object] = { "start_date": date_to_str(start), "end_date": date_to_str(end), "source_rows": source_rows_total, "loaded_rows": 0, "main_appended_rows": 0, "main_append_null_ohlc_skipped_rows": 0, "batches": [], } if dry_run: return result loaded_total = 0 with tempfile.TemporaryDirectory(prefix=f"kline_{period}_{run_id}_") as tmp: tmp_dir = Path(tmp) if period == "1m": minute_trade_dates = source_minute_trade_dates(source_conn, start, end, symbols) minute_symbols = symbols if symbols else source_symbols_from_daily(source_conn, start, end) symbol_groups = list(chunked(minute_symbols, minute_symbols_per_batch)) if minute_symbols else [[]] result["minute_trade_day_count"] = len(minute_trade_dates) result["minute_symbol_count"] = len(minute_symbols) result["minute_symbols_per_batch"] = minute_symbols_per_batch result["batch_log_truncated_count"] = 0 for idx, (batch_start_dt, batch_end_dt, symbol_group_index, symbol_group) in enumerate( ( (window[0], window[1], symbol_group_index, symbol_group) for trade_day in minute_trade_dates for window in market_datetime_batches(trade_day, minute_max_minutes) for symbol_group_index, symbol_group in enumerate(symbol_groups, start=1) ), start=1, ): batch_info = { "batch_index": idx, "start_ts": batch_start_dt.strftime("%Y-%m-%d %H:%M:%S"), "end_ts": batch_end_dt.strftime("%Y-%m-%d %H:%M:%S"), "symbol_batch_index": symbol_group_index, "symbol_count": len(symbol_group), "source_rows": 0, "loaded_rows": 0, } tsv_path = tmp_dir / f"{period}_{idx:04d}.tsv" export_tsv(source_conn, build_minute_select_window(batch_start_dt, batch_end_dt, symbol_group), tsv_path) batch_source_rows = count_tsv_rows(tsv_path) batch_info["source_rows"] = batch_source_rows if batch_source_rows == 0: if len(result["batches"]) < 500: result["batches"].append(batch_info) else: result["batch_log_truncated_count"] = int(result["batch_log_truncated_count"]) + 1 continue load_minute(local_conn, tsv_path, run_id) # Symbol-batched windows are disjoint within a run; avoid an # increasingly expensive per-window recount after every symbol # chunk. batch_loaded_rows = batch_source_rows loaded_total += batch_loaded_rows result["source_rows"] = int(result["source_rows"]) + batch_source_rows if not coverage_rows else result["source_rows"] batch_info["loaded_rows"] = batch_loaded_rows if append_to_main: append_result = append_minute_window_to_main(local_conn, batch_start_dt, batch_end_dt, run_id, symbol_group) batch_info.update( { "main_appended_rows": append_result["appended_rows"], "main_append_null_ohlc_skipped_rows": append_result["null_ohlc_skipped_rows"], } ) result["main_appended_rows"] = int(result["main_appended_rows"]) + append_result["appended_rows"] result["main_append_null_ohlc_skipped_rows"] = int( result["main_append_null_ohlc_skipped_rows"] ) + append_result["null_ohlc_skipped_rows"] if len(result["batches"]) < 500: result["batches"].append(batch_info) else: result["batch_log_truncated_count"] = int(result["batch_log_truncated_count"]) + 1 result["loaded_rows"] = loaded_total return result for idx, (batch_start, batch_end) in enumerate(date_batches(start, end, max_days), start=1): batch_source_rows = count_source_rows(source_conn, period, batch_start, batch_end, symbols) batch_info: dict[str, object] = { "batch_index": idx, "start_date": date_to_str(batch_start), "end_date": date_to_str(batch_end), "source_rows": batch_source_rows, "loaded_rows": 0, } if batch_source_rows == 0: result["batches"].append(batch_info) continue tsv_path = tmp_dir / f"{period}_{idx:04d}.tsv" select_sql = build_daily_select(batch_start, batch_end, symbols) if period == "1d" else build_minute_select( batch_start, batch_end, symbols ) export_tsv(source_conn, select_sql, tsv_path) if period == "1d": load_daily(local_conn, tsv_path, run_id) else: load_minute(local_conn, tsv_path, run_id) batch_loaded_rows = local_loaded_count(local_conn, period, batch_start, batch_end, run_id) loaded_total += batch_loaded_rows batch_info["loaded_rows"] = batch_loaded_rows if append_to_main and period == "1d": append_result = append_daily_to_main(local_conn, batch_start, batch_end, run_id) batch_info.update( { "main_appended_rows": append_result["appended_rows"], "main_append_null_ohlc_skipped_rows": append_result["null_ohlc_skipped_rows"], } ) result["main_appended_rows"] = int(result["main_appended_rows"]) + append_result["appended_rows"] result["main_append_null_ohlc_skipped_rows"] = int( result["main_append_null_ohlc_skipped_rows"] ) + append_result["null_ohlc_skipped_rows"] result["batches"].append(batch_info) result["loaded_rows"] = loaded_total return result def build_connections(args: argparse.Namespace) -> tuple[MysqlConn, MysqlConn]: mysql_exe = Path(args.mysql_exe) source = MysqlConn( mysql_exe=mysql_exe, host=args.source_host, port=str(args.source_port), user=args.source_user, password=args.source_password or os.environ.get("KLINE_SOURCE_MYSQL_PASSWORD"), database=args.source_database, ) local = MysqlConn( mysql_exe=mysql_exe, host=args.local_host, port=str(args.local_port), user=args.local_user, password=args.local_password or os.environ.get("TIANXIA_MYSQL_PASSWORD"), database=args.local_database, ) if not source.password: raise RuntimeError("source password is required: pass --source-password or set KLINE_SOURCE_MYSQL_PASSWORD") if not local.password: raise RuntimeError("local password is required: pass --local-password or set TIANXIA_MYSQL_PASSWORD") return source, local def parse_symbols(raw: str | None) -> list[str]: if not raw: return [] parts: list[str] = [] for item in raw.replace(";", ",").split(","): item = item.strip() if item: parts.append(item) return parts def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--mysql-exe", default=str(DEFAULT_MYSQL_EXE)) parser.add_argument("--run-id", default=f"kline_front_incremental_sync_{datetime.now().strftime('%Y%m%d_%H%M%S')}") parser.add_argument("--periods", nargs="+", choices=["1d", "1m"], default=["1d", "1m"]) parser.add_argument("--start-date") parser.add_argument("--end-date") parser.add_argument("--symbols", help="comma separated symbols; omit for all A-share rows in the date range") parser.add_argument("--daily-max-days-per-batch", type=int, default=31) parser.add_argument("--minute-max-days-per-batch", type=int, default=3) parser.add_argument("--minute-max-minutes-per-batch", type=int, default=60) parser.add_argument("--minute-symbols-per-batch", type=int, default=50) parser.add_argument( "--append-to-main", action="store_true", help="append successfully synced mirror rows into local main tables without updating existing rows", ) parser.add_argument( "--allow-partial-main-append", action="store_true", help="allow --append-to-main when --symbols is provided; otherwise main append requires all-A sync", ) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--init-schema-only", action="store_true") parser.add_argument("--source-host", default=os.environ.get("KLINE_SOURCE_MYSQL_HOST", "317w7246e5.vicp.fun")) parser.add_argument("--source-port", default=os.environ.get("KLINE_SOURCE_MYSQL_PORT", "50176")) parser.add_argument("--source-user", default=os.environ.get("KLINE_SOURCE_MYSQL_USER", "root")) parser.add_argument("--source-password") parser.add_argument("--source-database", default=os.environ.get("KLINE_SOURCE_MYSQL_DB", "trading_xuntou")) parser.add_argument("--local-host", default=os.environ.get("TIANXIA_MYSQL_HOST", "127.0.0.1")) parser.add_argument("--local-port", default=os.environ.get("TIANXIA_MYSQL_PORT", "3306")) parser.add_argument("--local-user", default=os.environ.get("TIANXIA_MYSQL_USER", "root")) parser.add_argument("--local-password") parser.add_argument("--local-database", default=os.environ.get("TIANXIA_MYSQL_DB", "tianxia")) return parser.parse_args() def main() -> None: args = parse_args() symbols = parse_symbols(args.symbols) if args.append_to_main and symbols and not args.allow_partial_main_append: raise RuntimeError( "--append-to-main with --symbols would create a partial-date main table. " "Use --allow-partial-main-append only for explicit repair tasks." ) source_conn, local_conn = build_connections(args) create_local_schema(local_conn) if args.init_schema_only: print(f"schema_ready=1 run_id={args.run_id}") return run_start = datetime.now() insert_run_log_start(local_conn, args.run_id, args.periods, None, None, symbols, source_conn) summary: dict[str, object] = { "run_id": args.run_id, "started_at": run_start.isoformat(timespec="seconds"), "dry_run": int(args.dry_run), "source": { "host": source_conn.host, "port": source_conn.port, "database": source_conn.database, "tables": {"1d": SOURCE_DAILY_TABLE, "1m": SOURCE_MINUTE_TABLE}, "write_policy": "READ_ONLY_SOURCE_NO_HTTP_BACKFILL", }, "local": { "host": local_conn.host, "port": local_conn.port, "database": local_conn.database, "tables": { "1d": LOCAL_DAILY_TABLE, "1m": LOCAL_MINUTE_TABLE, "run_log": LOCAL_RUN_TABLE, "main_1d": MAIN_DAILY_TABLE, "main_1m": MAIN_MINUTE_TABLE, }, "write_policy": "MIRROR_TABLES_PLUS_APPEND_ONLY_MAIN" if args.append_to_main else "MIRROR_TABLES_ONLY", }, "symbols_scope": symbols if symbols else "ALL_A", "periods": {}, } status = "SUCCESS" error_message = None try: resolved_starts: list[date] = [] resolved_ends: list[date] = [] for period in args.periods: start = resolve_start_date(local_conn, period, args.start_date) end = resolve_end_date(source_conn, period, args.end_date) resolved_starts.append(start) resolved_ends.append(end) if start > end: summary["periods"][period] = { "start_date": date_to_str(start), "end_date": date_to_str(end), "source_rows": 0, "loaded_rows": 0, "status": "NOOP_LOCAL_ALREADY_UP_TO_DATE_OR_SOURCE_EMPTY", } continue max_days = args.daily_max_days_per_batch if period == "1d" else args.minute_max_days_per_batch period_result = sync_period( source_conn, local_conn, period, start, end, symbols, args.run_id, max_days, args.minute_max_minutes_per_batch, args.minute_symbols_per_batch, args.dry_run, args.append_to_main and not args.dry_run, ) period_result["status"] = "DRY_RUN" if args.dry_run else "SYNCED" summary["periods"][period] = period_result summary["resolved_start_date"] = date_to_str(min(resolved_starts)) if resolved_starts else "" summary["resolved_end_date"] = date_to_str(max(resolved_ends)) if resolved_ends else "" except Exception as exc: status = "FAILED" error_message = str(exc) summary["error_message"] = error_message raise finally: summary["finished_at"] = datetime.now().isoformat(timespec="seconds") try: update_run_log_finish(local_conn, args.run_id, status, summary, error_message) except Exception: if status != "FAILED": raise out_dir = RESULT_ROOT / args.run_id out_dir.mkdir(parents=True, exist_ok=True) (out_dir / "summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()