diff --git a/scripts/strategy_lifecycle.py b/scripts/strategy_lifecycle.py index eacf9247..12dce8e9 100644 --- a/scripts/strategy_lifecycle.py +++ b/scripts/strategy_lifecycle.py @@ -898,6 +898,7 @@ def reassess_strategy(code, name, price, cost, shares, current_action, sr = tech["support_resistance"] candle = tech.get("candlestick", {}) vol = tech.get("volume", {}) + vol_deep = tech.get("volume_deep", {}) ss = sr.get("strong_support") ws = sr.get("weak_support") wr = sr.get("weak_resist") @@ -1268,6 +1269,25 @@ def reassess_strategy(code, name, price, cost, shares, current_action, # 融合大盘趋势、行业板块强弱、基本面估值作为修正因子 volume_signal = vol.get("volume_signal", "") candlestick_sentiment = candle.get("sentiment", "neutral") + + # ----- 深度量价分析(2026-07-08 新增:放量建仓/出货/洗盘检测)----- + deep_vol = {} + vol_pattern_note = "" + try: + from technical_analysis import analyze_volume_deep + deep_vol = analyze_volume_deep(code) + if deep_vol.get("volume_signal"): + vp = deep_vol.get("price_vol_description", "") + patterns = deep_vol.get("patterns", {}) + vol_pattern_note = vp + # 合并到volume_signal + if volume_signal: + volume_signal += f" | {vp}" if vp else "" + else: + volume_signal = vp or "" + except Exception: + pass + timing_signal = "neutral" # --- 三维分析数据装载 --- @@ -1438,6 +1458,10 @@ def reassess_strategy(code, name, price, cost, shares, current_action, if action_note: action_parts.append(action_note) + # 量价分析描述(如有,注入action) + if vol_pattern_note: + action_parts.append(vol_pattern_note) + if is_watchlist: # 自选股(未入场):有止损参考+买入区,内部算RR需要止盈位 action_parts.append(f"目标参考{new_target}") diff --git a/scripts/technical_analysis.py b/scripts/technical_analysis.py index b5128578..27426959 100644 --- a/scripts/technical_analysis.py +++ b/scripts/technical_analysis.py @@ -116,6 +116,8 @@ def get_quote(code): h = get(33) # high l = get(34) # low c = get(3) # price / close + v = get(6) # volume(手) + amt = get(37) # 成交额 if h and l and c: history = _load_history() if raw not in history: @@ -126,8 +128,13 @@ def get_quote(code): days[-1]["high"] = max(days[-1]["high"], h) days[-1]["low"] = min(days[-1]["low"], l) days[-1]["close"] = c # 盘中用最新价,收盘后是收盘价 + if v: days[-1]["volume"] = v + if amt: days[-1]["amount"] = amt else: - days.append({"date": today_str, "high": h, "low": l, "close": c}) + entry = {"date": today_str, "high": h, "low": l, "close": c} + if v: entry["volume"] = v + if amt: entry["amount"] = amt + days.append(entry) # 只保留最近 HISTORY_DAYS 天 history[raw] = days[-HISTORY_DAYS:] _save_history(history) @@ -348,6 +355,240 @@ def analyze_volume(q): return result +def analyze_volume_trend(code): + """量价趋势分析:对比历史N日平均成交量,检测量价背离模式 + + 从 price_history.json 读取历史数据,比较今日量价关系。 + """ + result = {} + try: + history = _load_history() + days = history.get(code, []) + if len(days) < 3: + result["trend"] = "数据不足" + return result + + today = days[-1] + prev = days[-2] if len(days) >= 2 else None + today_vol = today.get("volume", 0) + today_close = today.get("close", 0) + + if not today_vol or not today_close: + result["trend"] = "数据不足" + return result + + # 计算N日均量 + vols_5 = [d.get("volume", 0) for d in days[-6:-1] if d.get("volume")] + vols_20 = [d.get("volume", 0) for d in days[-21:-1] if d.get("volume")] + avg_5 = sum(vols_5) / len(vols_5) if vols_5 else 0 + avg_20 = sum(vols_20) / len(vols_20) if vols_20 else 0 + + vol_ratio_vs_5 = today_vol / avg_5 if avg_5 > 0 else 0 + vol_ratio_vs_20 = today_vol / avg_20 if avg_20 > 0 else 0 + + result["avg_volume_5d"] = round(avg_5, 0) + result["avg_volume_20d"] = round(avg_20, 0) + result["today_volume"] = int(today_vol) + result["volume_ratio_vs_5d"] = round(vol_ratio_vs_5, 2) + result["volume_ratio_vs_20d"] = round(vol_ratio_vs_20, 2) + + # 最近3日的收盘价和成交量趋势 + if len(days) >= 3: + recent_close = [d.get("close", 0) for d in days[-4:-1]] + recent_vol = [d.get("volume", 0) for d in days[-4:-1]] + if all(recent_close) and all(recent_vol): + price_up = today_close > recent_close[-1] + vol_up = today_vol > recent_vol[-1] + + # 量价模式判定 + if vol_ratio_vs_5 >= 1.8: + # 明显放量 + if price_up: + result["trend"] = "放量上攻" + result["action"] = "buy_conformation" + else: + # 价格下跌但大幅放量 = 恐慌?还是承接收筹? + # 看今日K线:如果是阳线(低开高走)= 承接 + # 简单判断:如果close > open = 有承接 + result["trend"] = "放量下跌" + result["action"] = "watch" + elif vol_ratio_vs_5 <= 0.6: + # 明显缩量 + if price_up: + result["trend"] = "缩量上涨" + result["action"] = "divergence" + else: + result["trend"] = "缩量回调" + result["action"] = "healthy_pullback" + elif vol_ratio_vs_5 >= 1.3: + # 温和放量 + if price_up: + result["trend"] = "温和放量上涨" + result["action"] = "bullish" + else: + result["trend"] = "温和放量下跌" + result["action"] = "bearish" + else: + # 正常量 + if price_up: + result["trend"] = "正常量上涨" + result["action"] = "neutral_bullish" + else: + result["trend"] = "正常量下跌" + result["action"] = "neutral_bearish" + + # 量价背离检测:价格走高但成交量逐日递减 + if len(days) >= 5: + close_5 = [d.get("close", 0) for d in days[-5:]] + vol_5 = [d.get("volume", 0) for d in days[-5:]] + if all(close_5) and all(vol_5): + close_trend = close_5[-1] - close_5[0] + vol_trend = vol_5[-1] - vol_5[0] + # 价格涨但量跌 = 顶背离 + if close_trend > 0 and vol_trend < 0 and abs(vol_trend) > sum(vol_5) * 0.3: + result["divergence"] = "顶背离(价涨量缩)" + # 价格跌但量涨 = 底背离 + elif close_trend < 0 and vol_trend > 0 and abs(vol_trend) > sum(vol_5) * 0.3: + result["divergence"] = "底背离(价跌量增)" + + except Exception as e: + result["trend_error"] = str(e) + + return result + + +def analyze_volume_deep(code): + """深度量价分析:从日K线分析量价配合/背离/建仓/出货 + + 使用 mtf_cache 表的日K线数据做历史量价分析。 + """ + import sqlite3 + from pathlib import Path + + DATA_DIR = Path(__file__).parent / "data" + try: + conn = sqlite3.connect(str(DATA_DIR / "mofin.db")) + row = conn.execute("SELECT cache_json FROM mtf_cache WHERE code=?", (code,)).fetchone() + conn.close() + if not row: + return {"volume_signal": "数据不足"} + data = json.loads(row[0]) + except Exception: + return {"volume_signal": "数据不足"} + + daily = data.get("daily", []) + if len(daily) < 5: + return {"volume_signal": "数据不足"} + + closes = [d["close"] for d in daily] + volume = [d["volume"] for d in daily] + n = len(daily) + + # 基准:最近20日均量(不足20日则用全部) + lookback = min(20, n - 1) + avg_vol_20d = sum(volume[-lookback-1:-1]) / lookback if lookback > 0 else volume[-1] + + # 最近N日的量比 + recent = min(5, n) + recent_vol_ratios = [] + for i in range(recent): + vol = volume[-i-1] if i+1 <= n else volume[0] + recent_vol_ratios.append(round(vol / avg_vol_20d, 2) if avg_vol_20d > 0 else 1) + + today_ratio = recent_vol_ratios[0] if recent_vol_ratios else 1 + recent_max_ratio = max(recent_vol_ratios) if recent_vol_ratios else 1 + + # 量价配合度 + signals = [] + patterns = {} + + # 1. 放量检测(量比 > 2x) + if today_ratio > 2.0: + signals.append(f"量比{today_ratio:.1f}倍放量") + patterns["volume_surge"] = True + # 放量方向 + if len(closes) >= 2 and closes[-1] > closes[-2]: + patterns["surge_direction"] = "放量上涨" + if today_ratio > 2.5 and closes[-1] > closes[-2] * 1.03: + signals[-1] += "↑主力买入" + else: + signals[-1] += "↑" + elif len(closes) >= 2 and closes[-1] < closes[-2]: + patterns["surge_direction"] = "放量下跌" + if today_ratio > 2.5 and closes[-1] < closes[-2] * 0.97: + signals[-1] += "↓主力出货" + else: + signals[-1] += "↓" + else: + patterns["surge_direction"] = "放量平盘" + elif today_ratio < 0.5: + signals.append(f"量比{today_ratio:.1f}倍缩量") + patterns["volume_shrink"] = True + else: + signals.append(f"量比{today_ratio:.1f}倍正常") + patterns["volume_normal"] = True + + # 2. 量价趋势分析(近5日 vs 前5日) + if len(daily) >= 10: + recent5_vol = sum(volume[-5:]) / 5 + prev5_vol = sum(volume[-10:-5]) / 5 + vol_trend = "增" if recent5_vol > prev5_vol * 1.3 else ("减" if recent5_vol < prev5_vol * 0.7 else "稳") + recent5_price = closes[-5:] + price_trend = "涨" if recent5_price[-1] > recent5_price[0] else ("跌" if recent5_price[-1] < recent5_price[0] * 0.95 else "平") + + if vol_trend == "增" and price_trend == "涨": + patterns["accumulation"] = True # 量价齐升=建仓 + signals.append(f"近5日{vol_trend}量{price_trend}价=建仓型") + elif vol_trend == "增" and price_trend == "跌": + patterns["distribution"] = True # 放量下跌=出货 + signals.append(f"近5日{vol_trend}量{price_trend}价=⚠️出货型") + elif vol_trend == "减" and price_trend == "涨": + patterns["divergence"] = True # 量缩价涨=背离 + signals.append(f"近5日{vol_trend}量{price_trend}价=⬆量价背离") + elif vol_trend == "减" and price_trend == "跌": + patterns["washout"] = True # 缩量下跌=洗盘末端 + signals.append(f"近5日{vol_trend}量{price_trend}价=洗盘特征") + else: + signals.append(f"近5日{vol_trend}量{price_trend}价") + else: + vol_trend = price_trend = "?" + + # 3. 寻找历史放量区间(主力活动痕迹) + surge_days = [] + for i in range(max(0, n - 60), n): + vol_ratio = volume[i] / avg_vol_20d if avg_vol_20d > 0 else 0 + if vol_ratio > 2.0: + surge_days.append({ + "date": daily[i].get("date", ""), + "ratio": round(vol_ratio, 1), + "close": closes[i], + "direction": "涨" if (i > 0 and closes[i] > closes[i-1]) else "跌" + }) + + # 汇总描述 + vol_level = "放量" if today_ratio > 2.0 else ("缩量" if today_ratio < 0.5 else "正常") + price_vol = f"{vol_level}" + if patterns.get("accumulation"): + price_vol = f"量价齐升(建仓特征) | {signals[-1]}" + elif patterns.get("distribution"): + price_vol = f"放量下跌⚠️ | {signals[-1]}" + elif patterns.get("washout"): + price_vol = f"缩量回踩(洗盘末端) | {signals[-1]}" + elif patterns.get("divergence"): + price_vol = f"量价背离 | {signals[-1]}" + + return { + "volume_signal": " ; ".join(signals) if signals else "正常", + "volume_ratio": today_ratio, + "avg_volume_20d": int(avg_vol_20d), + "recent_ratios": recent_vol_ratios, + "surge_count_60d": len(surge_days), + "price_vol_description": price_vol, + "patterns": patterns, + "surge_days": surge_days[-5:] if surge_days else [], + } + + def full_analysis(code): """完整技术分析(带30秒缓存,避免分钟级波动)""" import time @@ -364,6 +605,12 @@ def full_analysis(code): sr = calc_support_resistance(q) candle = analyze_candlestick(q) vol = analyze_volume(q) + # 深度量价分析(使用日K线历史数据) + vol_deep = {} + try: + vol_deep = analyze_volume_deep(code) + except Exception: + pass # graceful degradation # 多周期+均线分析(整合 multi_timeframe) mtf = {} @@ -403,6 +650,7 @@ def full_analysis(code): "support_resistance": sr, "candlestick": candle, "volume": vol, + "volume_deep": vol_deep, "multi_tf": mtf, "analyzed_at": datetime.now().strftime("%H:%M"), }