新增逻辑脚本

2026-04-24 16:59:59 +08:00
parent 8827ba8cd6
commit f843197e27
4 changed files with 112 additions and 6 deletions
@@ -0,0 +1,106 @@
+import pandas as pd
+from sqlalchemy import create_engine
+
+
+def analyze_stock_performance(start_date, end_date):
+    # --- 数据库配置 ---
+    try:
+        engine = create_engine(
+            "mysql+pymysql://yangfan@8.159.129.156:10836/level?charset=utf8mb4",
+            connect_args={
+                "password": "aA%8023321088"
+            },
+            pool_pre_ping=True
+        )
+
+        # 1. 获取涨停当日股票及价格
+        limit_query = f"""
+        SELECT l.ts_code, l.name, d.open AS limit_day_open, d.high AS limit_day_high
+        FROM stock_limit_data l
+        LEFT JOIN stock_daily d ON l.ts_code = d.ts_code AND l.trade_date = d.trade_date
+        WHERE l.trade_date = '{start_date}' AND l.status = 1
+        """
+        df_limit = pd.read_sql(limit_query, engine)
+
+        if df_limit.empty:
+            print(f"提示：{start_date} 当日没有涨停记录。")
+            return
+
+        # 2. 获取行情区间数据
+        codes = df_limit['ts_code'].unique().tolist()
+        codes_tuple = f"('{codes[0]}')" if len(codes) == 1 else str(tuple(codes))
+        daily_query = f"""
+        SELECT ts_code, trade_date, high, low, close
+        FROM stock_daily
+        WHERE ts_code IN {codes_tuple} AND trade_date BETWEEN '{start_date}' AND '{end_date}'
+        ORDER BY trade_date ASC
+        """
+        df_daily = pd.read_sql(daily_query, engine)
+        df_daily['trade_date'] = pd.to_datetime(df_daily['trade_date'])
+
+        # 3. 核心逻辑处理
+        final_report = []
+        for _, limit_row in df_limit.iterrows():
+            code = limit_row['ts_code']
+            stock_data = df_daily[df_daily['ts_code'] == code].copy()
+
+            if stock_data.empty: continue
+
+            # A. 找到区间内的最高价及其第一次出现的日期
+            max_high = float(stock_data['high'].max())
+            # 获取最高价所在的那一行（如果有多个相同最高价，取第一个）
+            max_high_row = stock_data[stock_data['high'] == max_high].iloc[0]
+            max_date = max_high_row['trade_date']
+
+            # B. 寻找最高价日期之后的最低价
+            post_max_data = stock_data[stock_data['trade_date'] > max_date]
+
+            if not post_max_data.empty:
+                min_low = float(post_max_data['low'].min())
+                last_close = float(post_max_data.iloc[-1]['close'])
+
+                price_diff = max_high - min_low
+                drop_pct = (price_diff / max_high * 100)
+                rebound_pct = ((last_close - min_low) / min_low * 100) if min_low > 0 else 0
+            else:
+                # 如果最高点出现在结束日当天，则没有“后期最低价”
+                min_low = None
+                price_diff = 0
+                drop_pct = 0
+                rebound_pct = 0
+
+            final_report.append({
+                "代码": code,
+                "名称": limit_row['name'],
+                "当日开盘": float(limit_row['limit_day_open'] or 0),
+                "当日最高": float(limit_row['limit_day_high'] or 0),
+                "区间最高": max_high,
+                "最高日期": max_date.strftime('%Y-%m-%d'),
+                "见顶后最低": min_low,
+                "最大回落幅%": round(drop_pct, 2),
+                "低位至期末波动%": round(rebound_pct, 2)
+            })
+
+        # 4. 排序与输出
+        if final_report:
+            result_df = pd.DataFrame(final_report)
+            result_df = result_df.sort_values(by="最大回落幅%", ascending=False)
+
+            print(f"\n分析区间: {start_date} 至 {end_date}")
+            print("说明: '见顶后最低' 指最高价日期之后的最低点")
+            print("=" * 120)
+            print(result_df.to_string(index=False, na_rep='N/A'))
+            print("=" * 120)
+
+    except Exception as e:
+        print(f"发生错误: {e}")
+
+
+# ==========================================
+# 设定日期
+# ==========================================
+if __name__ == "__main__":
+    START_INPUT = '2024-09-24'
+    END_INPUT = '2026-04-23'
+
+    analyze_stock_performance(START_INPUT, END_INPUT)