yang/python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增 akshare 逻辑（从 2026-04-01开始）

Browse Source
This commit is contained in:
mianyang
2026-04-30 00:07:48 +08:00
parent 5e8abfd2c2
commit a2b80e9d85
1 changed files with 242 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
limit_up.py
+242 -26
View File
@@ -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()