Qionk/a-share-quant

1"""2斐波那契回调分析3计算关键水平并生成交易信号4"""5 6import numpy as np7import pandas as pd8 9FIB_LEVELS = [0.0, 0.236, 0.382, 0.5, 0.618, 0.786, 1.0, 1.272, 1.618]10FIB_NAMES = {11    0.0: "低点", 0.236: "23.6%", 0.382: "38.2%", 0.5: "50.0%",12    0.618: "61.8%", 0.786: "78.6%", 1.0: "高点",13    1.272: "127.2%", 1.618: "161.8%",14}15FIB_COLORS = {16    0.0: "blue", 0.236: "orange", 0.382: "orange",17    0.5: "gray", 0.618: "green", 0.786: "green",18    1.0: "blue", 1.272: "purple", 1.618: "purple",19}20 21 22def calculate_fibonacci_levels(df: pd.DataFrame, lookback_days: int = 90) -> dict:23    """24    从最近 lookback_days 天内计算斐波那契回调水平。25 26    返回: {27        "levels": {level: price, ...},28        "swing_high": float,29        "swing_low": float,30        "range": float,31        "trend": "up" | "down",32    }33    """34    recent = df.tail(lookback_days)35    swing_high = recent["high"].max()36    swing_low = recent["low"].min()37    swing_range = swing_high - swing_low38 39    if swing_range < swing_high * 0.01:40        swing_range = swing_high * 0.0141 42    mid_idx = lookback_days // 243    trend = "up" if recent["close"].iloc[-1] > recent["close"].iloc[min(mid_idx, len(recent) - 1)] else "down"44 45    levels = {}46    for level in FIB_LEVELS:47        levels[level] = swing_high - level * swing_range48 49    return {50        "levels": levels,51        "swing_high": swing_high,52        "swing_low": swing_low,53        "range": swing_range,54        "trend": trend,55    }56 57 58def generate_fibonacci_signals(current_price: float, fib_data: dict,59                                 predictions: dict = None) -> list:60    """61    基于斐波那契水平生成买卖信号。62 63    返回: [{"signal": "buy"/"sell", "level": str, "price": float, "description": str}, ...]64    """65    levels = fib_data["levels"]66    signals = []67    threshold = 0.0268 69    # 支撑位 -> 买入信号70    for level in [0.618, 0.786]:71        price = levels[level]72        if abs(current_price - price) / current_price < threshold:73            signals.append({74                "signal": "buy",75                "level": FIB_NAMES[level],76                "price": round(price, 2),77                "description": f"价格接近 {FIB_NAMES[level]} 支撑位 (¥{price:.2f})",78            })79 80    # 阻力位 -> 卖出信号81    for level in [0.236, 0.382]:82        price = levels[level]83        if abs(current_price - price) / current_price < threshold:84            signals.append({85                "signal": "sell",86                "level": FIB_NAMES[level],87                "price": round(price, 2),88                "description": f"价格接近 {FIB_NAMES[level]} 阻力位 (¥{price:.2f})",89            })90 91    # 与预测交叉检查92    if predictions and predictions.get("predicted_close") is not None \93       and len(predictions["predicted_close"]) > 0:94        pred_prices = predictions["predicted_close"]95        for level in [0.618, 0.786]:96            if any(abs(p - levels[level]) / max(p, 0.01) < 0.03 for p in pred_prices):97                signals.append({98                    "signal": "buy",99                    "level": FIB_NAMES[level],100                    "description": f"预测价格可能触及 {FIB_NAMES[level]} 支撑位",101                })102        for level in [0.236, 0.382]:103            if any(abs(p - levels[level]) / max(p, 0.01) < 0.03 for p in pred_prices):104                signals.append({105                    "signal": "sell",106                    "level": FIB_NAMES[level],107                    "description": f"预测价格可能触及 {FIB_NAMES[level]} 阻力位",108                })109 110    # 去重111    seen = set()112    unique = []113    for s in signals:114        if s["description"] not in seen:115            seen.add(s["description"])116            unique.append(s)117    return unique