diff --git a/cal_date.py b/cal_date.py index 05b68cd..4bdd287 100644 --- a/cal_date.py +++ b/cal_date.py @@ -131,7 +131,7 @@ def run(start_date, end_date, token): if __name__ == "__main__": parser = argparse.ArgumentParser(description="同步交易日历") parser.add_argument("--start-date", default="20260430", help="开始日期，格式 YYYYMMDD") - parser.add_argument("--end-date", default="20260430", help="结束日期，格式 YYYYMMDD") + parser.add_argument("--end-date", default="20260506", help="结束日期，格式 YYYYMMDD") args = parser.parse_args() run(args.start_date, args.end_date, TUSHARE_TOKEN) diff --git a/daily.py b/daily.py index 9018e67..001af22 100644 --- a/daily.py +++ b/daily.py @@ -158,7 +158,7 @@ def run(start_date, end_date): if __name__ == "__main__": parser = argparse.ArgumentParser(description="同步日行情") parser.add_argument("--start-date", default="20260430", help="开始日期，格式 YYYYMMDD") - parser.add_argument("--end-date", default="20260430", help="结束日期，格式 YYYYMMDD") + parser.add_argument("--end-date", default="20260506", help="结束日期，格式 YYYYMMDD") args = parser.parse_args() run(args.start_date, args.end_date) diff --git a/limit_up.py b/limit_up.py index 274a2f5..032eb65 100644 --- a/limit_up.py +++ b/limit_up.py @@ -49,13 +49,14 @@ def filter_dates_by_start(trade_dates, start_date, source_name): return filtered_dates -def init_db(engine): +def init_db(engine, full_maintenance=False): """ 初始化数据库表结构 """ with engine.begin() as conn: + print(f" - 确保 {COMBINED_TABLE} 存在 ...") conn.execute(text(get_combined_table_sql())) - optimize_combined_table(conn) + optimize_combined_table(conn, full_maintenance=full_maintenance) def get_combined_table_sql(): @@ -106,6 +107,22 @@ def get_table_columns(conn, table_name): } +def get_column_collation(conn, table_name, column_name): + row = conn.execute( + text( + """ + SELECT collation_name + FROM information_schema.columns + WHERE table_schema = DATABASE() + AND table_name = :table_name + AND column_name = :column_name + """ + ), + {"table_name": table_name, "column_name": column_name}, + ).first() + return row[0] if row else None + + def index_exists(conn, table_name, index_name): return conn.execute( text( @@ -166,6 +183,27 @@ def ensure_combined_columns(conn): ) ) + ensure_combined_join_collation(conn) + + +def ensure_combined_join_collation(conn): + daily_collation = get_column_collation(conn, "stock_daily", "ts_code") + combined_collation = get_column_collation(conn, COMBINED_TABLE, "ts_code") + if not daily_collation or not combined_collation or daily_collation == combined_collation: + return + + print(f" - 对齐 ts_code collation：{combined_collation} -> {daily_collation} ...") + conn.execute( + text( + f""" + ALTER TABLE `{COMBINED_TABLE}` + MODIFY `ts_code` VARCHAR(20) + CHARACTER SET utf8mb4 COLLATE {daily_collation} + NOT NULL COMMENT 'TS股票代码，以stock_daily为准' + """ + ) + ) + def sync_combined_vol_from_daily(conn): conn.execute( @@ -173,8 +211,8 @@ def sync_combined_vol_from_daily(conn): f""" UPDATE `{COMBINED_TABLE}` c JOIN `stock_daily` d - ON c.trade_date = d.trade_date - AND c.ts_code COLLATE utf8mb4_unicode_ci = d.ts_code COLLATE utf8mb4_unicode_ci + ON c.ts_code = d.ts_code + AND c.trade_date = d.trade_date SET c.symbol = d.symbol, c.calc_vol = d.vol """ @@ -189,8 +227,8 @@ def dedupe_combined_table(conn): DELETE c FROM `{COMBINED_TABLE}` c LEFT JOIN `stock_daily` d - ON c.trade_date = d.trade_date - AND c.ts_code COLLATE utf8mb4_unicode_ci = d.ts_code COLLATE utf8mb4_unicode_ci + ON c.ts_code = d.ts_code + AND c.trade_date = d.trade_date WHERE d.ts_code IS NULL """ ) @@ -202,7 +240,7 @@ def dedupe_combined_table(conn): FROM `{COMBINED_TABLE}` c1 JOIN `{COMBINED_TABLE}` c2 ON c1.trade_date = c2.trade_date - AND c1.ts_code COLLATE utf8mb4_unicode_ci = c2.ts_code COLLATE utf8mb4_unicode_ci + AND c1.ts_code = c2.ts_code WHERE ( COALESCE(c1.source_count, 0) * 10 @@ -236,16 +274,30 @@ def dedupe_combined_table(conn): ) -def ensure_combined_indexes(conn): - dedupe_combined_table(conn) +def ensure_combined_indexes(conn, full_maintenance=False): + needs_unique_rebuild = index_exists( + conn, COMBINED_TABLE, "uniq_date_code" + ) and not index_matches( + conn, COMBINED_TABLE, "uniq_date_code", ["trade_date", "ts_code"], unique=True + ) + needs_unique_create = not index_matches( + conn, COMBINED_TABLE, "uniq_date_code", ["trade_date", "ts_code"], unique=True + ) + + if full_maintenance or needs_unique_rebuild or needs_unique_create: + print(" - 清理历史重复/无效融合数据 ...") + dedupe_combined_table(conn) + if index_exists(conn, COMBINED_TABLE, "uniq_date_code") and not index_matches( conn, COMBINED_TABLE, "uniq_date_code", ["trade_date", "ts_code"], unique=True ): + print(" - 重建 uniq_date_code 唯一索引 ...") conn.execute(text(f"ALTER TABLE `{COMBINED_TABLE}` DROP INDEX `uniq_date_code`")) if not index_matches( conn, COMBINED_TABLE, "uniq_date_code", ["trade_date", "ts_code"], unique=True ): + print(" - 创建 uniq_date_code 唯一索引 ...") conn.execute( text( f"ALTER TABLE `{COMBINED_TABLE}` " @@ -253,14 +305,17 @@ def ensure_combined_indexes(conn): ) ) if not index_exists(conn, COMBINED_TABLE, "idx_trade_date"): + print(" - 创建 idx_trade_date 索引 ...") conn.execute( text(f"ALTER TABLE `{COMBINED_TABLE}` ADD KEY `idx_trade_date` (`trade_date`)") ) if not index_exists(conn, COMBINED_TABLE, "idx_symbol"): + print(" - 创建 idx_symbol 索引 ...") conn.execute( text(f"ALTER TABLE `{COMBINED_TABLE}` ADD KEY `idx_symbol` (`symbol`)") ) if not index_exists(conn, COMBINED_TABLE, "idx_trade_date_status"): + print(" - 创建 idx_trade_date_status 索引 ...") conn.execute( text( f"ALTER TABLE `{COMBINED_TABLE}` " @@ -269,10 +324,17 @@ def ensure_combined_indexes(conn): ) -def optimize_combined_table(conn): +def optimize_combined_table(conn, full_maintenance=False): + print(" - 检查字段 ...") ensure_combined_columns(conn) - sync_combined_vol_from_daily(conn) - ensure_combined_indexes(conn) + if full_maintenance: + print(" - 从 stock_daily 全量回填 calc_vol ...") + sync_combined_vol_from_daily(conn) + else: + print(" - 跳过全量 calc_vol 回填，当前批次写入时会直接带入 vol。") + print(" - 检查索引 ...") + ensure_combined_indexes(conn, full_maintenance=full_maintenance) + print(" - 校验重复键 ...") validate_no_duplicate_combined_rows(conn) @@ -711,7 +773,7 @@ def analyze_and_save_limits(start_date, end_date): def cleanup_combined_data(): engine = get_engine() try: - init_db(engine) + init_db(engine, full_maintenance=True) print(f"✅ {COMBINED_TABLE} 已完成历史重复数据清理、唯一索引检查和 calc_vol 回填。") return True except Exception as e: @@ -725,7 +787,7 @@ def cleanup_combined_data(): if __name__ == "__main__": parser = argparse.ArgumentParser(description="计算并保存涨停/炸板数据") parser.add_argument("--start-date", default="2026-04-30", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期") - parser.add_argument("--end-date", default="2026-04-30", help="结束日期，格式 YYYY-MM-DD") + parser.add_argument("--end-date", default="2026-05-06", help="结束日期，格式 YYYY-MM-DD") parser.add_argument( "--cleanup-only", action="store_true",