From a2b80e9d85b290816a80c2f8e94807855db60ec0 Mon Sep 17 00:00:00 2001 From: mianyang <784310601@qq.com> Date: Thu, 30 Apr 2026 00:07:48 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=20akshare=20=E9=80=BB?= =?UTF-8?q?=E8=BE=91=EF=BC=88=E4=BB=8E=202026-04-01=E5=BC=80=E5=A7=8B?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- limit_up.py | 268 +++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 242 insertions(+), 26 deletions(-) diff --git a/limit_up.py b/limit_up.py index 6174c54..0f22afe 100644 --- a/limit_up.py +++ b/limit_up.py @@ -1,19 +1,33 @@ 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): """ 初始化数据库表结构 - 1. 增加 symbol 列，位于 trade_date 之后 - 2. 建立 UNIQUE KEY 防止物理重复写入 """ create_table_sql = """ CREATE TABLE IF NOT EXISTS `stock_limit_data` ( @@ -26,11 +40,202 @@ def init_db(engine): `limit_price` DECIMAL(16, 4) COMMENT '计算出的理论涨停价', `close` DECIMAL(16, 4) 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_ak_columns(conn) + + +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", "market", + *ak_columns, + ]] def analyze_and_save_limits(start_date, end_date): @@ -43,19 +248,24 @@ def analyze_and_save_limits(start_date, end_date): init_db(engine) # 1. 获取基础数据和行情数据 - query = f""" + query = text(""" SELECT d.trade_date, d.symbol, d.ts_code, b.name, d.high, d.close, d.pre_close, 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}' + 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) + 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 @@ -81,30 +291,36 @@ def analyze_and_save_limits(start_date, end_date): save_df = pd.concat([limit_df, broken_df]) - if not save_df.empty: - # 4. 字段清洗与对齐 - final_df = save_df[[ - 'trade_date', 'symbol', 'ts_code', 'name', - 'status', 'limit_price', 'close', 'market' - ]] + target_dates = df['trade_date'].astype(str).unique() + ak_df = fetch_ak_limit_pool(target_dates) - # 5. 【核心防重逻辑】：开启事务，先删后插 - with engine.begin() as conn: - # 获取待写入的所有交易日期 - target_dates = final_df['trade_date'].astype(str).unique() - date_list = ", ".join([f"'{d}'" for d in 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) - # 第一层逻辑防重：删除数据库中已存在的相同日期数据 - delete_query = f"DELETE FROM stock_limit_data WHERE trade_date IN ({date_list})" - conn.execute(text(delete_query)) - - # 第二层物理防重：通过 to_sql 写入，若触发 Unique Key 冲突会抛出异常回滚 + if not save_df.empty: + final_df = save_df[[ + 'trade_date', 'symbol', 'ts_code', 'name', + 'status', 'limit_price', 'close', '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)} 条记录（已自动去重并覆盖历史数据）。") + print(f"📊 总计入库: {len(final_df)} 条记录，AKShare 匹配补充: {matched_count} 条。") else: - print("ℹ️ 该时段内全市场无涨停或炸板记录。") + print(f"ℹ️ 该时段内全市场无涨停或炸板记录，已清理 {len(target_dates)} 个交易日的旧结果。") return True except Exception as e: @@ -117,7 +333,7 @@ def analyze_and_save_limits(start_date, end_date): # ============================================================ if __name__ == "__main__": parser = argparse.ArgumentParser(description="计算并保存涨停/炸板数据") - parser.add_argument("--start-date", default="2026-04-29", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期") + parser.add_argument("--start-date", default="2026-03-31", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期") parser.add_argument("--end-date", default="2026-04-29", help="结束日期，格式 YYYY-MM-DD") args = parser.parse_args()