使用cci、kdj和站上5日线为买入标志

main.py

@@ -73,8 +73,11 @@ class TelegramQuantStateMachine(QuantStateMachine):
                 msg += "⚠️ *[变盘警告]*：弹簧已压紧，随时大突破，做T手速要快！\n"
             if f['percent_b'] >= 1.0 or f['min_bias'] > 0.035:
                 msg += f"🟢 *【建议高抛】*：当前处于震荡高位(价格:{f['close']})，建议尾盘或明日开盘*手动卖出网格仓*！"
-            elif f['percent_b'] <= 0.0 or f['min_bias'] < -0.035:
-                msg += f"🔴 *【建议低吸】*：当前处于震荡超跌区(价格:{f['close']})，建议尾盘或明日开盘*手动买回筹码*！"
+            # elif f['percent_b'] <= 0.0 or f['min_bias'] < -0.035:
+            elif ((f['cci'] > f['prev_cci'] and f['prev_cci'] > f['prev_cci_2'])  ## cci连续两天上涨
+                and (f['K'] > f['D'] and f['prev_K'] <= f['prev_D'])  ## kdj金叉
+                and (f['close'] > f['ma5'])): ## 收盘价站上五日线
+                msg += f"🔴 *【建议低吸】*：当前处于震荡超跌区(价格:{f['close']}，J：{f['J']})，建议尾盘或明日开盘*手动买回筹码*！"
             else:
                 msg += "⚪ *【建议观望】*：处于安全中枢内，未触及边界，明天*不要乱动*。"
         elif self.current_state == "STATE_3_MAIN_WAVE":

quant_strategy.py

@@ -1,7 +1,7 @@
 import numpy as np
 import pandas as pd
 import talib
-from util import calculate_brar
+from util import calculate_brar, calc_kdj_tdx
 
 # =====================================================================
 # 1. 指标引擎：未来所有新发掘的指标，全写在这里
@@ -60,6 +60,14 @@ class FeatureEngine:
 
         self.df["bb_up"], self.df["bb_mid"], self.df["bb_low"] = up, mid, low
 
+        df_tmp = calc_kdj_tdx(self.df)
+        self.df["K"] = df_tmp['K']
+        self.df["D"] = df_tmp['D']
+        self.df["J"] = df_tmp['J']
+
+        cci = talib.CCI(self.df['high'], self.df['low'], self.df['close'], timeperiod=14)
+        self.df["cci"] = cci
+
         df_tmp = calculate_brar(self.df)
 
         self.df["ar"] = df_tmp["ar"]
@@ -75,6 +83,7 @@ class FeatureEngine:
 
         latest = self.df.iloc[-1]
         prev = self.df.iloc[-2]
+        prev_2 = self.df.iloc[-3]
 
         # 如果 5分钟线数据拿到了，现价以 5分钟最新收盘价为准（更接近 14:48 真实盘面）
         # 如果没拿到，退化使用日线昨日收盘（做测试用）
@@ -96,11 +105,25 @@ class FeatureEngine:
         )
         features["is_not_new_low_10d"] = latest["close"] > latest["low_min_10d"]
 
+        features["prev_ma5"] = prev["ma5"] if "ma5" in prev else prev["close"]
+
+        features["K"] = latest["K"]
+        features["D"] = latest["D"]
+        features["J"] = latest["J"]
+
+        features["prev_K"] = prev["K"]
+        features["prev_D"] = prev["D"]
+        features["prev_J"] = prev["J"]
+
         features["ar"] = latest["ar"]
         features["prev_ar"] = prev["ar"]
         features["br"] = latest["br"]
         features["prev_br"] = prev["br"]
 
+        features["cci"] = latest["cci"]
+        features["prev_cci"] = prev["cci"]
+        features["prev_cci_2"] = prev_2["cci"]
+
         history_bw = self.df["bandwidth"].iloc[-250:]
         features["bw_quantile"] = (history_bw < latest["bandwidth"]).mean()
 

util.py

@@ -1,5 +1,6 @@
 import requests
 import pandas as pd
+import numpy as np
 
 CN_CODE_TYPE = {
     "600":'sh',
@@ -18,7 +19,7 @@ CN_CODE_TYPE = {
     "18":"sz"
     }
 
-
+## arbr指标
 def calculate_brar(df: pd.DataFrame, N: int = 26) -> pd.DataFrame:
     """
     计算AR和BR情绪指标
@@ -58,7 +59,37 @@ def calculate_brar(df: pd.DataFrame, N: int = 26) -> pd.DataFrame:
 
     return data
 
+## kdj计算
+def calc_kdj_tdx(df, n=9, m1=3, m2=3):
+    # 1. 计算 RSV
+    low_min = df['low'].rolling(n).min()
+    high_max = df['high'].rolling(n).max()
+    rsv = (df['close'] - low_min) / (high_max - low_min) * 100
+    rsv = rsv.fillna(0)  # 或者处理 NaN
 
+    # 2. 自定义 SMA 函数（通达信风格）
+    def sma_tdx(series, period, weight):
+        # period=N, weight=M。公式：Y = (X*M + PREV_Y*(N-M))/N
+        result = np.zeros(len(series))
+        # 初始化第一个有效值。通常，如果数据足够，初始值为 series[0]
+        result[0] = series[0] if not np.isnan(series[0]) else 0
+        for i in range(1, len(series)):
+            # 如果 series[i] 是 NaN（例如，前 n 天），则保持 0 或向前填充，但通常 RSV 已填充。
+            val = series[i] if not np.isnan(series[i]) else 0
+            result[i] = (val * weight + result[i - 1] * (period - weight)) / period
+        return result
+
+    # 3. 计算 K 和 D
+    k_series = sma_tdx(rsv.values, m1, 1)  # K = SMA(RSV, 3, 1)
+    d_series = sma_tdx(k_series, m2, 1)  # D = SMA(K, 3, 1)
+
+    df['K'] = k_series
+    df['D'] = d_series
+    df['J'] = 3 * df['K'] - 2 * df['D']
+
+    return df
+
+## 获取历史k线
 def get_history_k(code, start_date='19700101', end_date=None):
 
     if '.' in code:
@@ -93,6 +124,7 @@ def get_history_k(code, start_date='19700101', end_date=None):
     return df
 
 
+## 获取当日5分钟的k线
 def get_5min_k(code, start_date='19700101', end_date=None):
     stock_code = ''