"""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()
|
