import argparse from datetime import datetime from decimal import Decimal, ROUND_HALF_UP from sqlalchemy import text from stock_config import get_engine DETAIL_TABLE = "second_board_profit_detail_cache" SUMMARY_TABLE = "second_board_profit_summary_cache" BATCH_SIZE = 1000 def parse_date(value): value = str(value).strip() for fmt in ("%Y%m%d", "%Y-%m-%d"): try: return datetime.strptime(value, fmt).date() except ValueError: pass raise ValueError(f"日期格式错误: {value}，应为 YYYYMMDD 或 YYYY-MM-DD") def to_decimal(value): if value is None: return None return Decimal(str(value)) def round2(value): if value is None: return None return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) def pct_change(value, base): value = to_decimal(value) base = to_decimal(base) if value is None or base is None or base <= 0: return None return round2((value - base) / base * Decimal("100")) def format_limit_time(value): text_value = "" if value is None else str(value).strip() if not text_value or text_value.lower() in ("none", "null", "nan"): return "" if ":" in text_value: return text_value if len(text_value) == 4: text_value = f"{text_value}00" elif len(text_value) == 5: text_value = f"0{text_value}" elif len(text_value) != 6: return "" return f"{text_value[0:2]}:{text_value[2:4]}:{text_value[4:6]}" def time_seconds(value): text_value = format_limit_time(value) if not text_value: return None parts = text_value.split(":") if len(parts) < 2 or len(parts) > 3: return None try: hour = int(parts[0]) minute = int(parts[1]) second = int(parts[2]) if len(parts) == 3 else 0 except ValueError: return None if hour > 23 or minute > 59 or second > 59 or min(hour, minute, second) < 0: return None return hour * 3600 + minute * 60 + second def is_target_limit_time(value): seconds = time_seconds(value) if seconds is None: return False return 9 * 3600 + 31 * 60 <= seconds <= 9 * 3600 + 59 * 60 + 59 def init_db(engine): detail_sql = f""" CREATE TABLE IF NOT EXISTS `{DETAIL_TABLE}` ( `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '自增主键', `trade_date` DATE NOT NULL COMMENT 'T日交易日', `previous_trade_date` DATE NOT NULL COMMENT 'T-1交易日', `previous_two_trade_date` DATE NOT NULL COMMENT 'T-2交易日', `next_trade_date` DATE NULL COMMENT 'T+1交易日', `symbol` VARCHAR(10) NOT NULL DEFAULT '' COMMENT '股票代码', `ts_code` VARCHAR(20) NOT NULL COMMENT 'TuShare股票代码', `name` VARCHAR(50) NOT NULL DEFAULT '' COMMENT '股票简称', `market` VARCHAR(20) NOT NULL DEFAULT '' COMMENT '所属板块', `industry` VARCHAR(100) NOT NULL DEFAULT '' COMMENT '所属行业', `previous_close` DECIMAL(16,4) NULL COMMENT 'T-1收盘价', `limit_price` DECIMAL(16,4) NULL COMMENT 'T日涨停价', `t_high` DECIMAL(16,4) NULL COMMENT 'T日最高价', `t_close` DECIMAL(16,4) NULL COMMENT 'T日收盘价', `t_pct_chg` DECIMAL(10,4) NULL COMMENT 'T日涨跌幅', `status` VARCHAR(10) NOT NULL DEFAULT '' COMMENT 'limit封板 break炸板', `first_limit_time` VARCHAR(8) NOT NULL DEFAULT '' COMMENT 'T日首次涨停时间', `open_times` INT NULL COMMENT '开板次数', `next_high` DECIMAL(16,4) NULL COMMENT 'T+1最高价', `next_close` DECIMAL(16,4) NULL COMMENT 'T+1收盘价', `next_intraday_profitable` TINYINT(1) NULL COMMENT 'T+1盘中是否盈利', `next_high_profit_pct` DECIMAL(10,2) NULL COMMENT 'T+1最大获利百分比', `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `updated_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', PRIMARY KEY (`id`), UNIQUE KEY `uniq_trade_code` (`trade_date`, `ts_code`), KEY `idx_ts_code_date` (`ts_code`, `trade_date`), KEY `idx_trade_date` (`trade_date`), KEY `idx_total_profit` (`next_high_profit_pct`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ summary_sql = f""" CREATE TABLE IF NOT EXISTS `{SUMMARY_TABLE}` ( `ts_code` VARCHAR(20) NOT NULL COMMENT 'TuShare股票代码', `symbol` VARCHAR(10) NOT NULL DEFAULT '' COMMENT '股票代码', `name` VARCHAR(50) NOT NULL DEFAULT '' COMMENT '股票简称', `market` VARCHAR(20) NOT NULL DEFAULT '' COMMENT '所属板块', `industry` VARCHAR(100) NOT NULL DEFAULT '' COMMENT '所属行业', `second_board_count` INT NOT NULL DEFAULT 0 COMMENT '二板次数', `limit_count` INT NOT NULL DEFAULT 0 COMMENT '封板次数', `break_count` INT NOT NULL DEFAULT 0 COMMENT '炸板次数', `profit_data_count` INT NOT NULL DEFAULT 0 COMMENT '具备收益数据次数', `total_next_high_profit_pct` DECIMAL(12,2) NOT NULL DEFAULT 0 COMMENT '最大获利累计百分比', `average_next_high_profit_pct` DECIMAL(10,2) NOT NULL DEFAULT 0 COMMENT '最大获利平均百分比', `first_trade_date` DATE NULL COMMENT '首次命中日期', `last_trade_date` DATE NULL COMMENT '最近命中日期', `updated_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', PRIMARY KEY (`ts_code`), KEY `idx_total_profit` (`total_next_high_profit_pct`), KEY `idx_last_trade_date` (`last_trade_date`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ with engine.begin() as conn: conn.execute(text(detail_sql)) conn.execute(text(summary_sql)) print(f"✅ 表 {DETAIL_TABLE} / {SUMMARY_TABLE} 检查完成。") def load_candidate_rows(conn, start_date, end_date): query = text( """ WITH calendar AS ( SELECT cal_date AS trade_date, LAG(cal_date) OVER (ORDER BY cal_date) AS previous_trade_date, LAG(cal_date, 2) OVER (ORDER BY cal_date) AS previous_two_trade_date, LEAD(cal_date) OVER (ORDER BY cal_date) AS next_trade_date FROM trade_calendar WHERE exchange = 'SSE' AND is_open = 1 ), calendar_window AS ( SELECT * FROM calendar WHERE trade_date BETWEEN :start_date AND :end_date ) SELECT DATE_FORMAT(trade_date, '%Y-%m-%d') AS trade_date, DATE_FORMAT(previous_trade_date, '%Y-%m-%d') AS previous_trade_date, DATE_FORMAT(previous_two_trade_date, '%Y-%m-%d') AS previous_two_trade_date, COALESCE(DATE_FORMAT(next_trade_date, '%Y-%m-%d'), '') AS next_trade_date, COALESCE(symbol, '') AS symbol, ts_code, name, market, industry, previous_close, limit_price, t_high, t_close, t_pct_chg, status, first_limit_time, open_times, next_high, next_close FROM ( SELECT c.trade_date, c.previous_trade_date, c.previous_two_trade_date, c.next_trade_date, d.symbol, pl.ts_code, COALESCE(NULLIF(sb.name, ''), NULLIF(pl.name, ''), '') AS name, COALESCE(NULLIF(sb.market, ''), '') AS market, COALESCE(NULLIF(sb.industry, ''), NULLIF(pl.industry, ''), '') AS industry, COALESCE(d.pre_close, pl.close) AS previous_close, ROUND( COALESCE(d.pre_close, pl.close) * CASE WHEN COALESCE(sb.market, '') = '北交所' THEN 1.30 WHEN COALESCE(sb.market, '') IN ('创业板', '科创板') THEN 1.20 WHEN COALESCE(sb.name, pl.name, '') LIKE '%ST%' THEN 1.05 ELSE 1.10 END, 2 ) AS limit_price, d.high AS t_high, d.close AS t_close, d.pct_chg AS t_pct_chg, CASE WHEN ABS(d.close - d.high) < 0.0001 THEN 'limit' WHEN d.close < d.high THEN 'break' ELSE 'unknown' END AS status, CASE WHEN ABS(d.close - d.high) < 0.0001 THEN NULLIF(NULLIF(TRIM(tlu.first_time), ''), 'None') WHEN d.close < d.high THEN NULLIF(NULLIF(TRIM(tlz.first_time), ''), 'None') ELSE NULL END AS first_limit_time, CASE WHEN ABS(d.close - d.high) < 0.0001 THEN tlu.open_times WHEN d.close < d.high THEN tlz.open_times ELSE NULL END AS open_times, nd.high AS next_high, nd.close AS next_close FROM calendar_window c INNER JOIN limit_list_d pl ON pl.trade_date = c.previous_trade_date AND pl.`limit` = 'U' LEFT JOIN limit_list_d ppl ON ppl.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci AND ppl.trade_date = c.previous_two_trade_date AND ppl.`limit` = 'U' INNER JOIN stock_daily d ON d.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci AND d.trade_date = c.trade_date LEFT JOIN stock_daily nd ON nd.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci AND nd.trade_date = c.next_trade_date LEFT JOIN limit_list_d tlu ON tlu.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci AND tlu.trade_date = c.trade_date AND tlu.`limit` = 'U' LEFT JOIN limit_list_d tlz ON tlz.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci AND tlz.trade_date = c.trade_date AND tlz.`limit` = 'Z' LEFT JOIN stock_basic sb ON sb.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci WHERE c.previous_trade_date IS NOT NULL AND c.previous_two_trade_date IS NOT NULL AND ppl.ts_code IS NULL AND d.high IS NOT NULL AND d.close IS NOT NULL ) candidate WHERE previous_close IS NOT NULL AND limit_price IS NOT NULL AND limit_price > 0 AND t_high >= limit_price - 0.0001 AND status IN ('limit', 'break') ORDER BY trade_date ASC, ts_code ASC """ ) return [ dict(row) for row in conn.execute( query, {"start_date": start_date, "end_date": end_date}, ).mappings() ] def build_detail_records(rows): records = [] for row in rows: first_limit_time = format_limit_time(row["first_limit_time"]) if not is_target_limit_time(first_limit_time): continue next_intraday_profitable = None if row["next_high"] is not None and row["t_high"] is not None: next_intraday_profitable = ( 1 if to_decimal(row["next_high"]) > to_decimal(row["t_high"]) else 0 ) records.append( { "trade_date": row["trade_date"], "previous_trade_date": row["previous_trade_date"], "previous_two_trade_date": row["previous_two_trade_date"], "next_trade_date": row["next_trade_date"] or None, "symbol": row["symbol"] or "", "ts_code": row["ts_code"], "name": row["name"] or "", "market": row["market"] or "", "industry": row["industry"] or "", "previous_close": row["previous_close"], "limit_price": row["limit_price"], "t_high": row["t_high"], "t_close": row["t_close"], "t_pct_chg": row["t_pct_chg"], "status": row["status"], "first_limit_time": first_limit_time, "open_times": row["open_times"], "next_high": row["next_high"], "next_close": row["next_close"], "next_intraday_profitable": next_intraday_profitable, "next_high_profit_pct": pct_change(row["next_high"], row["t_high"]), } ) return records def load_existing_keys(conn, start_date, end_date): query = text( f""" SELECT DATE_FORMAT(trade_date, '%Y-%m-%d') AS trade_date, ts_code FROM `{DETAIL_TABLE}` WHERE trade_date BETWEEN :start_date AND :end_date """ ) rows = conn.execute( query, {"start_date": start_date, "end_date": end_date}, ).mappings() return {(row["trade_date"], row["ts_code"]) for row in rows} def write_detail_records(conn, records, start_date, end_date): existing_keys = load_existing_keys(conn, start_date, end_date) overwritten_count = sum( 1 for record in records if (record["trade_date"], record["ts_code"]) in existing_keys ) delete_result = conn.execute( text( f""" DELETE FROM `{DETAIL_TABLE}` WHERE trade_date BETWEEN :start_date AND :end_date """ ), {"start_date": start_date, "end_date": end_date}, ) if not records: return 0, overwritten_count, delete_result.rowcount sql = text( f""" INSERT INTO `{DETAIL_TABLE}` ( trade_date, previous_trade_date, previous_two_trade_date, next_trade_date, symbol, ts_code, name, market, industry, previous_close, limit_price, t_high, t_close, t_pct_chg, status, first_limit_time, open_times, next_high, next_close, next_intraday_profitable, next_high_profit_pct ) VALUES ( :trade_date, :previous_trade_date, :previous_two_trade_date, :next_trade_date, :symbol, :ts_code, :name, :market, :industry, :previous_close, :limit_price, :t_high, :t_close, :t_pct_chg, :status, :first_limit_time, :open_times, :next_high, :next_close, :next_intraday_profitable, :next_high_profit_pct ) ON DUPLICATE KEY UPDATE previous_trade_date = VALUES(previous_trade_date), previous_two_trade_date = VALUES(previous_two_trade_date), next_trade_date = VALUES(next_trade_date), symbol = VALUES(symbol), name = VALUES(name), market = VALUES(market), industry = VALUES(industry), previous_close = VALUES(previous_close), limit_price = VALUES(limit_price), t_high = VALUES(t_high), t_close = VALUES(t_close), t_pct_chg = VALUES(t_pct_chg), status = VALUES(status), first_limit_time = VALUES(first_limit_time), open_times = VALUES(open_times), next_high = VALUES(next_high), next_close = VALUES(next_close), next_intraday_profitable = VALUES(next_intraday_profitable), next_high_profit_pct = VALUES(next_high_profit_pct) """ ) for index in range(0, len(records), BATCH_SIZE): conn.execute(sql, records[index:index + BATCH_SIZE]) return len(records), overwritten_count, delete_result.rowcount def refresh_summary(conn): conn.execute(text(f"TRUNCATE TABLE `{SUMMARY_TABLE}`")) conn.execute( text( f""" INSERT INTO `{SUMMARY_TABLE}` ( ts_code, symbol, name, market, industry, second_board_count, limit_count, break_count, profit_data_count, total_next_high_profit_pct, average_next_high_profit_pct, first_trade_date, last_trade_date ) SELECT ts_code, MAX(symbol) AS symbol, MAX(name) AS name, MAX(market) AS market, MAX(industry) AS industry, COUNT(*) AS second_board_count, SUM(CASE WHEN status = 'limit' THEN 1 ELSE 0 END) AS limit_count, SUM(CASE WHEN status = 'break' THEN 1 ELSE 0 END) AS break_count, COUNT(next_high_profit_pct) AS profit_data_count, ROUND(SUM(COALESCE(next_high_profit_pct, 0)), 2) AS total_next_high_profit_pct, COALESCE( ROUND( SUM(COALESCE(next_high_profit_pct, 0)) / NULLIF(COUNT(next_high_profit_pct), 0), 2 ), 0 ) AS average_next_high_profit_pct, MIN(trade_date) AS first_trade_date, MAX(trade_date) AS last_trade_date FROM `{DETAIL_TABLE}` GROUP BY ts_code """ ) ) def run(start_date, end_date): start = parse_date(start_date) end = parse_date(end_date) if start > end: raise ValueError("开始日期不能大于结束日期") engine = get_engine() print(f"🚀 开始刷新二板获利缓存：{start} ~ {end}") try: init_db(engine) with engine.begin() as conn: rows = load_candidate_rows(conn, start, end) records = build_detail_records(rows) write_count, overwritten_count, deleted_count = write_detail_records( conn, records, start, end, ) refresh_summary(conn) print(f"候选记录：{len(rows)} 条") print(f"命中记录：{len(records)} 条") print(f"清理旧记录：{deleted_count} 条") print(f"写入记录：{write_count} 条") print(f"覆盖记录：{overwritten_count} 条") print("✅ 二板获利缓存刷新完成。") return True except Exception as exc: print(f"❌ 二板获利缓存刷新失败: {exc}") return False if __name__ == "__main__": parser = argparse.ArgumentParser(description="刷新二板获利缓存") parser.add_argument("--start-date", default="20260515", help="开始日期，格式 YYYYMMDD 或 YYYY-MM-DD") parser.add_argument("--end-date", default="20260515", help="结束日期，格式 YYYYMMDD 或 YYYY-MM-DD") args = parser.parse_args() run(args.start_date, args.end_date)