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

代码优化

Browse Source
This commit is contained in:
yangfan
2026-05-06 18:10:33 +08:00
parent 4f68ab8710
commit a914866c62
3 changed files with 78 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
limit_up.py
+76 -14
View File
@@ -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",