|
|
| """
|
| Wall Street Multi-dimension Quantitative Trading Model - Rule Engine
|
| Implements all 30 rules across 4 dimensions
|
| Rule return format: (triggered: bool, score: int, detail: dict)
|
| """
|
|
|
| import numpy as np
|
| import pandas as pd
|
|
|
|
|
|
|
|
|
|
|
|
|
| class VolumePriceRules:
|
| """VP-01 to VP-07 volume-price rules"""
|
|
|
| @staticmethod
|
| def vp01_breakout_with_volume(df: pd.Series, prev: pd.Series) -> tuple:
|
| """VP-01: breakout with volume confirmation buy (P0)"""
|
| score = 0
|
| detail = {"rule": "VP-01", "name": "volume_breakout", "triggered": False}
|
| volume_ma20 = df.get("volume_ma20", 0)
|
| if volume_ma20 is None or volume_ma20 == 0:
|
| return False, 0, detail
|
| vol_ratio = df["volume"] / volume_ma20 if volume_ma20 > 0 else 1
|
| vol_ok = vol_ratio > 1.5
|
| price_ok = df["close"] > df.get("high_20", df["close"])
|
| pct_ok = df.get("pct_chg", 0) > 3
|
| above_ma20 = df["close"] > df.get("ma20", df["close"])
|
| if vol_ok and price_ok and pct_ok and above_ma20:
|
| score = 20
|
| detail["triggered"] = True
|
| detail["vol_ratio"] = round(float(vol_ratio), 2)
|
| detail["pct_chg"] = round(float(df["pct_chg"]), 2)
|
| return detail["triggered"], score, detail
|
|
|
| @staticmethod
|
| def vp02_shrink_support(df: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| """VP-02: shrink volume back to support buy (P1)"""
|
| detail = {"rule": "VP-02", "name": "shrink_support", "triggered": False}
|
| if len(prev_N) < 20:
|
| return False, 0, detail
|
| vol_ma = df.get("volume_ma20", 1)
|
| if vol_ma == 0:
|
| return False, 0, detail
|
| recent_10 = prev_N.tail(10)
|
| has_volume_spike = any(recent_10["volume"] > 1.5 * vol_ma)
|
| if not has_volume_spike:
|
| return False, 0, detail
|
| vol_shrink = df["volume"] < 0.6 * vol_ma
|
| ma20 = df.get("ma20", 0)
|
| above_ma20 = df["close"] > ma20 * 0.98
|
| high_10 = recent_10["high"].max()
|
| pulled_back = (high_10 - df["close"]) / high_10 > 0.03
|
| if vol_shrink and above_ma20 and pulled_back:
|
| return True, 10, {
|
| **detail, "triggered": True,
|
| "vol_ratio_ma": round(float(df["volume"] / vol_ma), 2),
|
| "pullback_pct": round(float((high_10 - df["close"]) / high_10 * 100), 2)
|
| }
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def vp03_divergence_top(df: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| """VP-03: volume-price divergence top sell (P0)"""
|
| detail = {"rule": "VP-03", "name": "divergence_top", "triggered": False}
|
| if len(prev_N) < 30:
|
| return False, 0, detail
|
| recent = pd.concat([prev_N.tail(20), pd.DataFrame([df])], ignore_index=True)
|
| high_20 = recent["high"].max()
|
| is_new_high = df["close"] >= high_20 * 0.97 and df["close"] > prev_N.iloc[-1]["close"]
|
| vol_20_max = prev_N.tail(20)["volume"].max()
|
| vol_decreasing = df["volume"] < vol_20_max * 0.8
|
| macd_dif = df.get("macd_dif", 0)
|
| prev_dif = prev_N.tail(3)["macd_dif"].iloc[0] if len(prev_N) >= 3 else 0
|
| dif_decreasing = macd_dif < prev_dif
|
| if is_new_high and vol_decreasing and dif_decreasing:
|
| return True, -25, {
|
| **detail, "triggered": True,
|
| "new_high": round(float(df["close"]), 2),
|
| "vol_decreasing_pct": round(float(df["volume"] / vol_20_max * 100), 1)
|
| }
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def vp04_sky_volume_sky_price(df: pd.Series) -> tuple:
|
| """VP-04: sky volume sky price sell (P0)"""
|
| detail = {"rule": "VP-04", "name": "sky_volume", "triggered": False}
|
| vol_ma60 = df.get("volume_ma60", 0)
|
| if vol_ma60 is None or vol_ma60 == 0:
|
| return False, 0, detail
|
| vol_new_high = df["volume"] > vol_ma60 * 2.5
|
| turnover_high = df.get("turnover", 0) > 15
|
| body = abs(df["close"] - df["open"])
|
| upper_shadow = df["high"] - max(df["close"], df["open"])
|
| long_upper = upper_shadow > body * 2 and body > 0
|
| high_60 = df.get("high_60", df["high"])
|
| at_top = df["close"] > high_60 * 0.95
|
| if (vol_new_high or turnover_high) and (long_upper or df["close"] < df["open"]) and at_top:
|
| return True, -25, {**detail, "triggered": True,
|
| "vol_ratio_60": round(float(df["volume"] / vol_ma60), 2)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def vp05_extreme_shrink_bottom(df: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| """VP-05: extreme shrink bottom confirmation buy (P0)"""
|
| detail = {"rule": "VP-05", "name": "extreme_shrink", "triggered": False}
|
| if len(prev_N) < 60:
|
| return False, 0, detail
|
| vol_ma60 = df.get("volume_ma60", 0)
|
| if vol_ma60 is None or vol_ma60 == 0:
|
| return False, 0, detail
|
| vol_min_60 = prev_N.tail(60)["volume"].min()
|
| vol_min = df["volume"] <= vol_min_60 * 1.1
|
| recent_5 = prev_N.tail(5)
|
| low_volatility = (recent_5["high"].max() - recent_5["low"].min()) / recent_5["low"].min() < 0.05
|
| vol_increase = df["volume"] > df.get("volume_ma5", 0) * 2 and df["close"] > df["open"]
|
| if vol_min and low_volatility and vol_increase:
|
| return True, 20, {**detail, "triggered": True,
|
| "vol_ratio_60": round(float(df["volume"] / vol_ma60), 2)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def vp06_volume_surge(df: pd.Series, prev_1: pd.Series) -> tuple:
|
| """VP-06: volume ratio surge monitoring (P1)"""
|
| detail = {"rule": "VP-06", "name": "volume_surge", "triggered": False}
|
| vol_ratio = df.get("vol_ratio", 0)
|
| if vol_ratio is None:
|
| return False, 0, detail
|
| if vol_ratio > 2.0 and df["close"] > prev_1["close"] and df["close"] > df["open"]:
|
| return True, 10, {**detail, "triggered": True, "vol_ratio": round(float(vol_ratio), 2)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def vp07_price_up_vol_down(df: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| """VP-07: price up volume down divergence (P1)"""
|
| detail = {"rule": "VP-07", "name": "price_up_vol_down", "triggered": False}
|
| if len(prev_N) < 5:
|
| return False, 0, detail
|
| prices_up = all(prev_N.iloc[-i]["close"] > prev_N.iloc[-i-1]["close"] for i in range(1, 4))
|
| if not prices_up:
|
| return False, 0, detail
|
| vols = [prev_N.iloc[-i]["volume"] for i in range(1, 4)]
|
| vol_decreasing = all(vols[i] < vols[i-1] for i in range(1, len(vols)))
|
| cum_return = (df["close"] - prev_N.iloc[-4]["close"]) / prev_N.iloc[-4]["close"]
|
| return_ok = cum_return < 0.15
|
| j_value = df.get("kdj_j", 0)
|
| j_high = j_value > 100 if j_value is not None else False
|
| if vol_decreasing and return_ok and j_high:
|
| return True, -12, {**detail, "triggered": True,
|
| "cum_return": round(float(cum_return * 100), 2),
|
| "j_value": round(float(j_value), 1)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def evaluate_all(df_row: pd.Series, history: pd.DataFrame) -> dict:
|
| """Evaluate all VP rules"""
|
| results = {}
|
| score = 0
|
| prev = history.iloc[-2] if len(history) >= 2 else history.iloc[-1]
|
| triggered, s, d = VolumePriceRules.vp01_breakout_with_volume(df_row, prev)
|
| results["VP-01"] = d; score += s
|
| triggered, s, d = VolumePriceRules.vp02_shrink_support(df_row, history.tail(20))
|
| results["VP-02"] = d; score += s
|
| triggered, s, d = VolumePriceRules.vp03_divergence_top(df_row, history.tail(30))
|
| results["VP-03"] = d; score += s
|
| triggered, s, d = VolumePriceRules.vp04_sky_volume_sky_price(df_row)
|
| results["VP-04"] = d; score += s
|
| triggered, s, d = VolumePriceRules.vp05_extreme_shrink_bottom(df_row, history.tail(60))
|
| results["VP-05"] = d; score += s
|
| prev_1 = history.iloc[-2] if len(history) >= 2 else history.iloc[-1]
|
| triggered, s, d = VolumePriceRules.vp06_volume_surge(df_row, prev_1)
|
| results["VP-06"] = d; score += s
|
| triggered, s, d = VolumePriceRules.vp07_price_up_vol_down(df_row, history.tail(10))
|
| results["VP-07"] = d; score += s
|
| if results.get("VP-03", {}).get("triggered") or results.get("VP-04", {}).get("triggered"):
|
| dim_score = 0
|
| else:
|
| dim_score = max(30 + score, 0)
|
| return {"rules": results, "dim_score": min(dim_score, 100), "raw_score": score}
|
|
|
|
|
|
|
|
|
|
|
|
|
| class KCandlestickRules:
|
| """KC-01 to KC-08 K-line patterns"""
|
|
|
| @staticmethod
|
| def is_bullish(s): return s["close"] > s["open"]
|
| @staticmethod
|
| def is_bearish(s): return s["close"] < s["open"]
|
| @staticmethod
|
| def body_len(s): return abs(s["close"] - s["open"])
|
| @staticmethod
|
| def total_len(s): return s["high"] - s["low"]
|
|
|
| @staticmethod
|
| def kc01_morning_star(df_row: pd.Series, prev_2: pd.DataFrame) -> tuple:
|
| """KC-01: Morning star pattern (P0 buy)"""
|
| detail = {"rule": "KC-01", "name": "morning_star", "triggered": False}
|
| if len(prev_2) < 2: return False, 0, detail
|
| d1, d2 = prev_2.iloc[-2], prev_2.iloc[-1]
|
| if not KCandlestickRules.is_bearish(d1): return False, 0, detail
|
| body1 = KCandlestickRules.body_len(d1)
|
| if d1["high"] - d1["low"] == 0 or body1 / (d1["high"] - d1["low"]) < 0.3: return False, 0, detail
|
| gap_down = d2["open"] < d1["close"]
|
| if not KCandlestickRules.is_bullish(df_row): return False, 0, detail
|
| b3 = KCandlestickRules.body_len(df_row)
|
| t3 = KCandlestickRules.total_len(df_row)
|
| if t3 == 0 or b3 / t3 < 0.3: return False, 0, detail
|
| mid1 = (d1["high"] + d1["low"]) / 2
|
| above_mid = df_row["close"] > mid1
|
| vol_ok = df_row["volume"] > d2["volume"] * 1.5
|
| low_30 = prev_2.tail(30)["low"].min() if len(prev_2) >= 30 else d1["low"]
|
| at_bottom = d1["low"] <= low_30 * 1.05
|
| if gap_down and above_mid and vol_ok and at_bottom:
|
| return True, 20, {**detail, "triggered": True}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def kc02_bullish_engulfing(df_row: pd.Series, prev_1: pd.Series) -> tuple:
|
| """KC-02: Bullish engulfing (P0 buy)"""
|
| detail = {"rule": "KC-02", "name": "bullish_engulfing", "triggered": False}
|
| if not KCandlestickRules.is_bullish(df_row): return False, 0, detail
|
| bc = KCandlestickRules.body_len(df_row)
|
| tc = KCandlestickRules.total_len(df_row)
|
| if tc == 0 or bc / tc < 0.3: return False, 0, detail
|
| if df_row["open"] < prev_1["open"] and df_row["close"] > prev_1["close"]:
|
| vol_ok = df_row["volume"] > df_row.get("volume_ma20", df_row["volume"]) * 1.2
|
| if vol_ok:
|
| return True, 20, {**detail, "triggered": True}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def kc03_bearish_engulfing(df_row: pd.Series, prev_1: pd.Series) -> tuple:
|
| """KC-03: Bearish engulfing (P0 sell)"""
|
| detail = {"rule": "KC-03", "name": "bearish_engulfing", "triggered": False}
|
| if not KCandlestickRules.is_bearish(df_row): return False, 0, detail
|
| bc = KCandlestickRules.body_len(df_row)
|
| tc = KCandlestickRules.total_len(df_row)
|
| if tc == 0 or bc / tc < 0.3: return False, 0, detail
|
| if df_row["open"] > prev_1["open"] and df_row["close"] < prev_1["close"]:
|
| vol_ok = df_row["volume"] > df_row.get("volume_ma20", df_row["volume"]) * 1.2
|
| if vol_ok:
|
| return True, -25, {**detail, "triggered": True}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def kc04_bullish_harami(df_row: pd.Series, prev_1: pd.Series) -> tuple:
|
| """KC-04: Bullish harami (P1 buy)"""
|
| detail = {"rule": "KC-04", "name": "bullish_harami", "triggered": False}
|
| if not KCandlestickRules.is_bearish(prev_1): return False, 0, detail
|
| bp = KCandlestickRules.body_len(prev_1)
|
| tp = KCandlestickRules.total_len(prev_1)
|
| if tp == 0 or bp / tp < 0.3: return False, 0, detail
|
| if not KCandlestickRules.is_bullish(df_row): return False, 0, detail
|
| if df_row["high"] < prev_1["high"] and df_row["low"] > prev_1["low"]:
|
| return True, 10, {**detail, "triggered": True}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def kc05_rising_three_methods(df_row: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| """KC-05: Rising three methods (P1 buy)"""
|
| detail = {"rule": "KC-05", "name": "rising_three", "triggered": False}
|
| if len(prev_N) < 5: return False, 0, detail
|
| r5 = prev_N.tail(5)
|
| if not KCandlestickRules.is_bullish(r5.iloc[0]): return False, 0, detail
|
| for i in range(1, 4):
|
| if not KCandlestickRules.is_bearish(r5.iloc[i]): return False, 0, detail
|
| if r5.iloc[i]["low"] < r5.iloc[0]["low"]: return False, 0, detail
|
| if not KCandlestickRules.is_bullish(df_row): return False, 0, detail
|
| if df_row["close"] > r5["high"].max() and df_row["volume"] > df_row.get("volume_ma20", 1) * 1.2:
|
| return True, 10, {**detail, "triggered": True}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def kc06_evening_star(df_row: pd.Series, prev_2: pd.DataFrame) -> tuple:
|
| """KC-06: Evening star (P0 sell)"""
|
| detail = {"rule": "KC-06", "name": "evening_star", "triggered": False}
|
| if len(prev_2) < 2: return False, 0, detail
|
| d1, d2 = prev_2.iloc[-2], prev_2.iloc[-1]
|
| if not KCandlestickRules.is_bullish(d1): return False, 0, detail
|
| b2 = KCandlestickRules.body_len(d2)
|
| t2 = KCandlestickRules.total_len(d2)
|
| if t2 == 0 or b2 / t2 > 0.5: return False, 0, detail
|
| if not KCandlestickRules.is_bearish(df_row): return False, 0, detail
|
| mid1 = (d1["high"] + d1["low"]) / 2
|
| if df_row["close"] < mid1 and df_row["volume"] > d2["volume"] * 1.3:
|
| return True, -25, {**detail, "triggered": True}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def kc07_long_upper_shadow(df_row: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| """KC-07: Long upper shadow (P1)"""
|
| detail = {"rule": "KC-07", "name": "long_upper_shadow", "triggered": False}
|
| body = KCandlestickRules.body_len(df_row)
|
| upper = df_row["high"] - max(df_row["close"], df_row["open"])
|
| total = KCandlestickRules.total_len(df_row)
|
| if total == 0 or body == 0: return False, 0, detail
|
| if upper > body * 2:
|
| high_30 = prev_N.tail(30)["high"].max() if len(prev_N) >= 30 else df_row["high"]
|
| at_high = df_row["high"] >= high_30 * 0.95
|
| if at_high:
|
| if df_row["volume"] > df_row.get("volume_ma20", 1) * 1.3:
|
| return True, -12, {**detail, "triggered": True, "type": "distribution"}
|
| else:
|
| if df_row["volume"] < df_row.get("volume_ma20", 1) * 1.2:
|
| return True, 10, {**detail, "triggered": True, "type": "testing"}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def kc08_hammer_hanging(df_row: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| """KC-08: Hammer / Hanging man (P0)"""
|
| detail = {"rule": "KC-08", "name": "hammer_hanging", "triggered": False}
|
| body = KCandlestickRules.body_len(df_row)
|
| lower = min(df_row["close"], df_row["open"]) - df_row["low"]
|
| upper = df_row["high"] - max(df_row["close"], df_row["open"])
|
| total = KCandlestickRules.total_len(df_row)
|
| if total == 0 or body == 0: return False, 0, detail
|
| lr = lower / body
|
| if 2 <= lr <= 4 and upper < body * 0.3:
|
| low_30 = prev_N.tail(30)["low"].min() if len(prev_N) >= 30 else df_row["low"]
|
| high_30 = prev_N.tail(30)["high"].max() if len(prev_N) >= 30 else df_row["high"]
|
| pos = (df_row["close"] - low_30) / (high_30 - low_30) if high_30 > low_30 else 0.5
|
| if pos < 0.3:
|
| return True, 20, {**detail, "triggered": True, "type": "hammer_buy"}
|
| elif pos > 0.7 and df_row["volume"] > df_row.get("volume_ma20", 1) * 1.2:
|
| return True, -25, {**detail, "triggered": True, "type": "hanging_sell"}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def evaluate_all(df_row: pd.Series, history: pd.DataFrame) -> dict:
|
| results = {}; score = 0
|
| triggered, s, d = KCandlestickRules.kc01_morning_star(df_row, history.tail(10))
|
| results["KC-01"] = d; score += s
|
| prev_1 = history.iloc[-2] if len(history) >= 2 else history.iloc[-1]
|
| triggered, s, d = KCandlestickRules.kc02_bullish_engulfing(df_row, prev_1)
|
| results["KC-02"] = d; score += s
|
| triggered, s, d = KCandlestickRules.kc03_bearish_engulfing(df_row, prev_1)
|
| results["KC-03"] = d; score += s
|
| triggered, s, d = KCandlestickRules.kc04_bullish_harami(df_row, prev_1)
|
| results["KC-04"] = d; score += s
|
| triggered, s, d = KCandlestickRules.kc05_rising_three_methods(df_row, history.tail(10))
|
| results["KC-05"] = d; score += s
|
| triggered, s, d = KCandlestickRules.kc06_evening_star(df_row, history.tail(10))
|
| results["KC-06"] = d; score += s
|
| triggered, s, d = KCandlestickRules.kc07_long_upper_shadow(df_row, history.tail(30))
|
| results["KC-07"] = d; score += s
|
| triggered, s, d = KCandlestickRules.kc08_hammer_hanging(df_row, history.tail(30))
|
| results["KC-08"] = d; score += s
|
| if results.get("KC-03", {}).get("triggered") or results.get("KC-06", {}).get("triggered"):
|
| dim_score = 0
|
| else:
|
| dim_score = max(30 + score, 0)
|
| return {"rules": results, "dim_score": min(dim_score, 100), "raw_score": score}
|
|
|
|
|
|
|
|
|
|
|
|
|
| class ChipDistributionRules:
|
|
|
| @staticmethod
|
| def cp01_bottom_single_peak(df_row: pd.Series) -> tuple:
|
| detail = {"rule": "CP-01", "name": "bottom_peak", "triggered": False}
|
| conc = df_row.get("chip_concentration", 0)
|
| pr = df_row.get("profit_ratio", 0.5)
|
| if conc > 0.65 and pr < 0.3:
|
| return True, 20, {**detail, "triggered": True, "concentration": round(float(conc), 3), "profit_ratio": round(float(pr), 3)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def cp02_chip_upward_dispersion(df_row: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| detail = {"rule": "CP-02", "name": "chip_dispersion", "triggered": False}
|
| pr = df_row.get("profit_ratio", 0.5)
|
| if pr > 0.85 and df_row.get("pct_chg", 0) < 2:
|
| return True, -25, {**detail, "triggered": True, "profit_ratio": round(float(pr), 3)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def cp03_double_peak_breakout(df_row: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| detail = {"rule": "CP-03", "name": "double_peak", "triggered": False}
|
| pr = df_row.get("profit_ratio", 0.5)
|
| high_20 = df_row.get("high_20", df_row["close"])
|
| if 0.4 <= pr <= 0.6 and df_row["close"] >= high_20 * 0.99:
|
| return True, 10, {**detail, "triggered": True, "profit_ratio": round(float(pr), 3)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def cp04_profit_ratio_threshold(df_row: pd.Series) -> tuple:
|
| detail = {"rule": "CP-04", "name": "profit_ratio", "triggered": False}
|
| pr = df_row.get("profit_ratio", 0.5)
|
| if pr < 0.1:
|
| return True, 10, {**detail, "triggered": True, "signal": "oversold_buy", "profit_ratio": round(float(pr), 3)}
|
| elif pr >= 0.85:
|
| return True, -12, {**detail, "triggered": True, "signal": "overbought_sell", "profit_ratio": round(float(pr), 3)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def cp05_main_fund_holding(df_row: pd.Series) -> tuple:
|
| detail = {"rule": "CP-05", "name": "fund_holding", "triggered": False}
|
| pr = df_row.get("profit_ratio", 0.5)
|
| conc = df_row.get("chip_concentration", 0)
|
| if pr < 0.35 and conc > 0.6:
|
| return True, 10, {**detail, "triggered": True, "concentration": round(float(conc), 3)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def cp06_rapid_dispersion(df_row: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| detail = {"rule": "CP-06", "name": "rapid_dispersion", "triggered": False}
|
| if len(prev_N) < 5: return False, 0, detail
|
| conc_now = df_row.get("chip_concentration", 0.5)
|
| conc_5d = prev_N.iloc[-5].get("chip_concentration", 0.5)
|
| if conc_5d - conc_now > 0.15 and df_row.get("turnover", 0) > 8:
|
| return True, -25, {**detail, "triggered": True, "conc_drop": round(float(conc_5d - conc_now), 3)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def evaluate_all(df_row: pd.Series, history: pd.DataFrame) -> dict:
|
| results = {}; score = 0
|
| triggered, s, d = ChipDistributionRules.cp01_bottom_single_peak(df_row)
|
| results["CP-01"] = d; score += s
|
| triggered, s, d = ChipDistributionRules.cp02_chip_upward_dispersion(df_row, history)
|
| results["CP-02"] = d; score += s
|
| triggered, s, d = ChipDistributionRules.cp03_double_peak_breakout(df_row, history)
|
| results["CP-03"] = d; score += s
|
| triggered, s, d = ChipDistributionRules.cp04_profit_ratio_threshold(df_row)
|
| results["CP-04"] = d; score += s
|
| triggered, s, d = ChipDistributionRules.cp05_main_fund_holding(df_row)
|
| results["CP-05"] = d; score += s
|
| triggered, s, d = ChipDistributionRules.cp06_rapid_dispersion(df_row, history)
|
| results["CP-06"] = d; score += s
|
| if results.get("CP-02", {}).get("triggered") or results.get("CP-06", {}).get("triggered"):
|
| dim_score = 0
|
| else:
|
| dim_score = max(30 + score, 0)
|
| return {"rules": results, "dim_score": min(dim_score, 100), "raw_score": score}
|
|
|
|
|
|
|
|
|
|
|
|
|
| class MoneyFlowRules:
|
|
|
| @staticmethod
|
| def mf01_big_order_inflow(df_row: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| detail = {"rule": "MF-01", "name": "big_order_inflow", "triggered": False}
|
| ni = df_row.get("fund_net_inflow", 0)
|
| if ni is None: return False, 0, detail
|
| if len(prev_N) >= 3:
|
| pf = [prev_N.iloc[-i].get("fund_net_inflow", 0) for i in range(1, 4)]
|
| if all(f is not None and f > 0 for f in pf) and ni > 0:
|
| return True, 20, {**detail, "triggered": True, "total_inflow": round(float(sum(pf) + ni), 0)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def mf02_ddx_three_red(df_row: pd.Series) -> tuple:
|
| detail = {"rule": "MF-02", "name": "fund_three_red", "triggered": False}
|
| ni = df_row.get("fund_net_inflow", 0)
|
| nr = df_row.get("fund_net_inflow_ratio", 0)
|
| if ni is not None and nr is not None and ni > 500000 and nr > 0.05:
|
| return True, 20, {**detail, "triggered": True, "net_inflow": round(float(ni), 0)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def mf03_ddy_divergence(df_row: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| detail = {"rule": "MF-03", "name": "fund_price_divergence", "triggered": False}
|
| ni = df_row.get("fund_net_inflow", 0)
|
| if ni is None: return False, 0, detail
|
| if df_row.get("pct_chg", 0) < -1 and ni > 0:
|
| return True, 10, {**detail, "triggered": True, "net_inflow": round(float(ni), 0)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def mf04_main_force_retreat(df_row: pd.Series, prev_N: pd.DataFrame) -> tuple:
|
| detail = {"rule": "MF-04", "name": "main_force_retreat", "triggered": False}
|
| ni = df_row.get("fund_net_inflow", 0)
|
| if ni is None or len(prev_N) < 3: return False, 0, detail
|
| po = [prev_N.iloc[-i].get("fund_net_inflow", 0) for i in range(1, 3)]
|
| if all(f is not None and f < -500000 for f in po) and ni < -500000:
|
| return True, -25, {**detail, "triggered": True}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def mf05_dragon_tiger_institution(df_row: pd.Series) -> tuple:
|
| detail = {"rule": "MF-05", "name": "institution_buy", "triggered": False}
|
| ni = df_row.get("fund_net_inflow", 0)
|
| nr = df_row.get("fund_net_inflow_ratio", 0)
|
| to = df_row.get("turnover", 0)
|
| if ni is not None and nr is not None and ni > 2000000 and nr > 0.15 and to > 5:
|
| return True, 20, {**detail, "triggered": True, "net_inflow": round(float(ni), 0)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def mf06_north_bound_inflow(df_row: pd.Series) -> tuple:
|
| detail = {"rule": "MF-06", "name": "north_bound_flow", "triggered": False}
|
| ni = df_row.get("fund_net_inflow", 0)
|
| nr = df_row.get("fund_net_inflow_ratio", 0)
|
| if ni is not None and nr is not None and ni > 1000000 and nr > 0.08:
|
| return True, 10, {**detail, "triggered": True, "net_inflow": round(float(ni), 0)}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def mf07_margin_surge(df_row: pd.Series) -> tuple:
|
| """
|
| MF-07: 两融余额异动监测（P2）
|
| 条件: 融资净买入 > 前5日均值×2（买入信号）
|
| 融券净卖出 > 前5日均值×2（卖出信号）
|
| 数据: margin_net_buy, margin_balance（需外部数据源接入后生效）
|
|
|
| Returns:
|
| (triggered, score, detail)
|
| score = +15 (买入) 或 -15 (卖出)
|
| """
|
| detail = {"rule": "MF-07", "name": "margin_surge", "triggered": False}
|
| margin_net_buy = df_row.get("margin_net_buy", 0)
|
| if margin_net_buy is None or margin_net_buy == 0:
|
| return False, 0, detail
|
| margin_ma5 = df_row.get("margin_ma5", 0)
|
| if margin_ma5 is not None and margin_ma5 > 0 and margin_net_buy > margin_ma5 * 2:
|
| return True, 15, {
|
| **detail, "triggered": True,
|
| "direction": "bullish",
|
| "margin_net_buy": round(float(margin_net_buy), 0),
|
| "margin_ma5": round(float(margin_ma5), 0),
|
| "surge_ratio": round(float(margin_net_buy / margin_ma5), 2),
|
| }
|
| margin_short = df_row.get("margin_short", 0)
|
| short_ma5 = df_row.get("margin_short_ma5", 0)
|
| if short_ma5 is not None and short_ma5 > 0 and margin_short > short_ma5 * 2:
|
| return True, -15, {
|
| **detail, "triggered": True,
|
| "direction": "bearish",
|
| "margin_short": round(float(margin_short), 0),
|
| "short_ma5": round(float(short_ma5), 0),
|
| }
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def mf08_intraday_flow_verify(df_row: pd.Series) -> tuple:
|
| detail = {"rule": "MF-08", "name": "intraday_flow", "triggered": False}
|
| ni = df_row.get("fund_net_inflow", 0)
|
| pc = df_row.get("pct_chg", 0)
|
| am = df_row.get("amplitude", 0)
|
| if ni is not None and pc is not None and ni > 500000 and pc > 2 and 2 < am < 8:
|
| return True, 10, {**detail, "triggered": True}
|
| return False, 0, detail
|
|
|
| @staticmethod
|
| def evaluate_all(df_row: pd.Series, history: pd.DataFrame) -> dict:
|
| results = {}; score = 0
|
| triggered, s, d = MoneyFlowRules.mf01_big_order_inflow(df_row, history)
|
| results["MF-01"] = d; score += s
|
| triggered, s, d = MoneyFlowRules.mf02_ddx_three_red(df_row)
|
| results["MF-02"] = d; score += s
|
| triggered, s, d = MoneyFlowRules.mf03_ddy_divergence(df_row, history)
|
| results["MF-03"] = d; score += s
|
| triggered, s, d = MoneyFlowRules.mf04_main_force_retreat(df_row, history)
|
| results["MF-04"] = d; score += s
|
| triggered, s, d = MoneyFlowRules.mf05_dragon_tiger_institution(df_row)
|
| results["MF-05"] = d; score += s
|
| triggered, s, d = MoneyFlowRules.mf06_north_bound_inflow(df_row)
|
| results["MF-06"] = d; score += s
|
| triggered, s, d = MoneyFlowRules.mf07_margin_surge(df_row)
|
| results["MF-07"] = d; score += s
|
| triggered, s, d = MoneyFlowRules.mf08_intraday_flow_verify(df_row)
|
| results["MF-08"] = d; score += s
|
| if results.get("MF-04", {}).get("triggered"):
|
| dim_score = 0
|
| else:
|
| dim_score = max(30 + score, 0)
|
| return {"rules": results, "dim_score": min(dim_score, 100), "raw_score": score}
|
|
|
