import pandas as pd from sqlalchemy import text import argparse import re from stock_config import get_engine # --- 数据库配置 --- # 请根据实际情况修改数据库连接信息 # DB_CONN_STR = 'mysql+pymysql://root:password@127.0.0.1:3306/stock_data' AK_START_DATE = pd.to_datetime("2026-04-01").date() def normalize_stock_name(value): if pd.isna(value): return "" text_value = str(value).upper() text_value = re.sub(r"\s+", "", text_value) text_value = re.sub(r"[\*＊·・.。,_，、/\\()（）\[\]【】\-—_]+", "", text_value) for marker in ["行权", "除权", "复权"]: text_value = text_value.replace(marker, "") return text_value def init_db(engine): """ 初始化数据库表结构 """ create_table_sql = """ CREATE TABLE IF NOT EXISTS `stock_limit_data` ( `id` BIGINT AUTO_INCREMENT PRIMARY KEY, `trade_date` DATE NOT NULL COMMENT '交易日期', `symbol` VARCHAR(10) NOT NULL COMMENT '股票代码', `ts_code` VARCHAR(20) NOT NULL COMMENT 'TS股票代码', `name` VARCHAR(50) NOT NULL COMMENT '股票名称', `status` TINYINT NOT NULL COMMENT '状态：1-涨停，2-炸板', `limit_price` DECIMAL(16, 4) COMMENT '计算出的理论涨停价', `close` DECIMAL(16, 4) COMMENT '实际收盘价', `vol` BIGINT COMMENT '成交量', `market` VARCHAR(20) COMMENT '所属板块', `ak_code` VARCHAR(10) COMMENT 'AKShare-代码', `ak_name` VARCHAR(50) COMMENT 'AKShare-名称', `ak_pct_chg` DECIMAL(10, 4) COMMENT 'AKShare-涨跌幅', `ak_latest_price` DECIMAL(16, 4) COMMENT 'AKShare-最新价', `ak_amount` DECIMAL(24, 4) COMMENT 'AKShare-成交额', `ak_float_market_value` DECIMAL(24, 4) COMMENT 'AKShare-流通市值', `ak_total_market_value` DECIMAL(24, 4) COMMENT 'AKShare-总市值', `ak_turnover_rate` DECIMAL(10, 4) COMMENT 'AKShare-换手率', `ak_seal_amount` DECIMAL(24, 4) COMMENT 'AKShare-封板资金', `ak_first_limit_time` VARCHAR(20) COMMENT 'AKShare-首次封板时间', `ak_last_limit_time` VARCHAR(20) COMMENT 'AKShare-最后封板时间', `ak_break_limit_count` INT COMMENT 'AKShare-炸板次数', `ak_limit_up_stat` VARCHAR(50) COMMENT 'AKShare-涨停统计', `ak_limit_up_days` INT COMMENT 'AKShare-连板数', `ak_industry` VARCHAR(100) COMMENT 'AKShare-所属行业', UNIQUE KEY `uniq_date_code` (`trade_date`, `ts_code`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ with engine.begin() as conn: conn.execute(text(create_table_sql)) ensure_base_columns(conn) ensure_ak_columns(conn) def ensure_base_columns(conn): existing_columns = { row[0] for row in conn.execute(text("SHOW COLUMNS FROM `stock_limit_data`")).fetchall() } base_columns = { "vol": "BIGINT COMMENT '成交量'", } for column_name, column_def in base_columns.items(): if column_name not in existing_columns: conn.execute(text(f"ALTER TABLE `stock_limit_data` ADD COLUMN `{column_name}` {column_def}")) def ensure_ak_columns(conn): existing_columns = { row[0] for row in conn.execute(text("SHOW COLUMNS FROM `stock_limit_data`")).fetchall() } ak_columns = { "ak_code": "VARCHAR(10) COMMENT 'AKShare-代码'", "ak_name": "VARCHAR(50) COMMENT 'AKShare-名称'", "ak_pct_chg": "DECIMAL(10, 4) COMMENT 'AKShare-涨跌幅'", "ak_latest_price": "DECIMAL(16, 4) COMMENT 'AKShare-最新价'", "ak_amount": "DECIMAL(24, 4) COMMENT 'AKShare-成交额'", "ak_float_market_value": "DECIMAL(24, 4) COMMENT 'AKShare-流通市值'", "ak_total_market_value": "DECIMAL(24, 4) COMMENT 'AKShare-总市值'", "ak_turnover_rate": "DECIMAL(10, 4) COMMENT 'AKShare-换手率'", "ak_seal_amount": "DECIMAL(24, 4) COMMENT 'AKShare-封板资金'", "ak_first_limit_time": "VARCHAR(20) COMMENT 'AKShare-首次封板时间'", "ak_last_limit_time": "VARCHAR(20) COMMENT 'AKShare-最后封板时间'", "ak_break_limit_count": "INT COMMENT 'AKShare-炸板次数'", "ak_limit_up_stat": "VARCHAR(50) COMMENT 'AKShare-涨停统计'", "ak_limit_up_days": "INT COMMENT 'AKShare-连板数'", "ak_industry": "VARCHAR(100) COMMENT 'AKShare-所属行业'", } for column_name, column_def in ak_columns.items(): if column_name not in existing_columns: conn.execute(text(f"ALTER TABLE `stock_limit_data` ADD COLUMN `{column_name}` {column_def}")) def fetch_ak_limit_pool(trade_dates): try: import akshare as ak except ImportError as exc: raise RuntimeError("缺少 akshare 依赖，请先安装: pip install akshare") from exc frames = [] for trade_date in trade_dates: trade_day = pd.to_datetime(trade_date).date() if trade_day < AK_START_DATE: continue ak_date = pd.to_datetime(trade_date).strftime("%Y%m%d") try: ak_df = ak.stock_zt_pool_em(date=ak_date) except Exception as exc: print(f"⚠️ AKShare 涨停池获取失败: {ak_date}, 错误: {exc}") continue if ak_df is None or ak_df.empty: continue normalized_df = normalize_ak_limit_pool(ak_df, trade_date) if not normalized_df.empty: frames.append(normalized_df) if not frames: return pd.DataFrame(columns=["trade_date", "symbol", "name"]) return pd.concat(frames, ignore_index=True) def normalize_ak_limit_pool(ak_df, trade_date): column_map = { "代码": "ak_code", "名称": "ak_name", "涨跌幅": "ak_pct_chg", "最新价": "ak_latest_price", "成交额": "ak_amount", "流通市值": "ak_float_market_value", "总市值": "ak_total_market_value", "换手率": "ak_turnover_rate", "封板资金": "ak_seal_amount", "首次封板时间": "ak_first_limit_time", "最后封板时间": "ak_last_limit_time", "炸板次数": "ak_break_limit_count", "涨停统计": "ak_limit_up_stat", "连板数": "ak_limit_up_days", "所属行业": "ak_industry", } missing_columns = [column for column in column_map if column not in ak_df.columns] if missing_columns: print(f"⚠️ AKShare 返回字段缺失，已跳过缺失字段: {', '.join(missing_columns)}") if "代码" not in ak_df.columns or "名称" not in ak_df.columns: return pd.DataFrame(columns=["trade_date", "symbol", "name"]) available_map = { source: target for source, target in column_map.items() if source in ak_df.columns } normalized_df = ak_df[list(available_map)].rename(columns=available_map).copy() normalized_df["trade_date"] = pd.to_datetime(trade_date).date() normalized_df["symbol"] = ( normalized_df["ak_code"].astype(str).str.extract(r"(\d+)", expand=False).str.zfill(6) ) normalized_df["name"] = normalized_df["ak_name"].astype(str).str.strip() normalized_df["normalized_name"] = normalized_df["name"].apply(normalize_stock_name) for column in [ "ak_pct_chg", "ak_latest_price", "ak_amount", "ak_float_market_value", "ak_total_market_value", "ak_turnover_rate", "ak_seal_amount", "ak_break_limit_count", "ak_limit_up_days", ]: if column in normalized_df.columns: normalized_df[column] = pd.to_numeric(normalized_df[column], errors="coerce") for column in [ "ak_code", "ak_name", "ak_first_limit_time", "ak_last_limit_time", "ak_limit_up_stat", "ak_industry", ]: if column in normalized_df.columns: normalized_df[column] = normalized_df[column].astype(str) return normalized_df def merge_ak_limit_pool(final_df, ak_df): ak_columns = [ "ak_code", "ak_name", "ak_pct_chg", "ak_latest_price", "ak_amount", "ak_float_market_value", "ak_total_market_value", "ak_turnover_rate", "ak_seal_amount", "ak_first_limit_time", "ak_last_limit_time", "ak_break_limit_count", "ak_limit_up_stat", "ak_limit_up_days", "ak_industry", ] for column in ak_columns: if column not in ak_df.columns: ak_df[column] = None final_df["normalized_name"] = final_df["name"].apply(normalize_stock_name) if "normalized_name" not in ak_df.columns: ak_df["normalized_name"] = ak_df["name"].apply(normalize_stock_name) merge_keys = ["trade_date", "symbol"] ak_merge_df = ak_df[ merge_keys + ["normalized_name"] + ak_columns ].drop_duplicates(subset=merge_keys) merged_df = final_df.merge( ak_merge_df, how="left", on=merge_keys, suffixes=("", "_ak_match"), ) name_mismatch = ( merged_df["ak_code"].notna() & merged_df["normalized_name_ak_match"].notna() & (merged_df["normalized_name"] != merged_df["normalized_name_ak_match"]) ) if name_mismatch.any(): print(f"⚠️ 有 {int(name_mismatch.sum())} 条记录代码一致但名称不一致，已按代码合并 AKShare 字段。") return merged_df[[ "trade_date", "symbol", "ts_code", "name", "status", "limit_price", "close", "vol", "market", *ak_columns, ]] def analyze_and_save_limits(start_date, end_date): """ 计算并在数据库中落地涨停、炸板详情 """ engine = get_engine() # 0. 初始化检查表结构 init_db(engine) # 1. 获取基础数据和行情数据 query = text(""" SELECT d.trade_date, d.symbol, d.ts_code, b.name, d.high, d.close, d.pre_close, d.vol, b.market FROM stock_daily d JOIN stock_basic b ON d.ts_code = b.ts_code WHERE d.trade_date > :start_date AND d.trade_date <= :end_date AND d.pre_close IS NOT NULL ORDER BY d.trade_date ASC """) try: df = pd.read_sql( query, engine, params={"start_date": start_date, "end_date": end_date}, ) if df.empty: print(f"⚠️ 在指定区间 ({start_date} -> {end_date}] 内未获取到数据。") return True # 2. 涨停价计算规则 def calc_limit(row): m, p = str(row['market']), float(row['pre_close']) # 主板10%，创业板/科创板20%，北交所30% ratio = 1.2 if ('创业板' in m or '科创板' in m) else (1.3 if '北交所' in m else 1.1) return round(p * ratio, 2) df['limit_price'] = df.apply(calc_limit, axis=1) # 3. 筛选涨停(1)和炸板(2) df['is_limit'] = (df['high'] == df['limit_price']) & (df['close'] == df['limit_price']) df['is_broken'] = (df['high'] == df['limit_price']) & (df['close'] < df['limit_price']) limit_df = df[df['is_limit']].copy() limit_df['status'] = 1 broken_df = df[df['is_broken']].copy() broken_df['status'] = 2 save_df = pd.concat([limit_df, broken_df]) target_dates = df['trade_date'].astype(str).unique() ak_df = fetch_ak_limit_pool(target_dates) with engine.begin() as conn: delete_params = { f"date_{idx}": value for idx, value in enumerate(target_dates) } placeholders = ", ".join(f":date_{idx}" for idx in range(len(target_dates))) delete_query = text( f"DELETE FROM stock_limit_data WHERE trade_date IN ({placeholders})" ) conn.execute(delete_query, delete_params) if not save_df.empty: final_df = save_df[[ 'trade_date', 'symbol', 'ts_code', 'name', 'status', 'limit_price', 'close', 'vol', 'market' ]].copy() final_df["trade_date"] = pd.to_datetime(final_df["trade_date"]).dt.date final_df["symbol"] = final_df["symbol"].astype(str).str.zfill(6) final_df["name"] = final_df["name"].astype(str).str.strip() final_df = merge_ak_limit_pool(final_df, ak_df) final_df.to_sql('stock_limit_data', conn, if_exists='append', index=False) if not save_df.empty: matched_count = int(final_df["ak_code"].notna().sum()) print(f"✅ 处理成功！涉及交易日: {len(target_dates)} 天") print(f"📊 总计入库: {len(final_df)} 条记录，AKShare 匹配补充: {matched_count} 条。") else: print(f"ℹ️ 该时段内全市场无涨停或炸板记录，已清理 {len(target_dates)} 个交易日的旧结果。") return True except Exception as e: print(f"❌ 运行失败: {e}") return False # ============================================================ # 用户配置区 # ============================================================ if __name__ == "__main__": parser = argparse.ArgumentParser(description="计算并保存涨停/炸板数据") parser.add_argument("--start-date", default="2026-04-29", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期") parser.add_argument("--end-date", default="2026-04-30", help="结束日期，格式 YYYY-MM-DD") args = parser.parse_args() # 开始日期：计算从该日期的下一个交易日开始 START_DATE = args.start_date # 结束日期 END_DATE = args.end_date analyze_and_save_limits(START_DATE, END_DATE)