hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 严谨技术指标(RSI/MACD/量能/背离)_calc_ta_score+_calc_scores

Browse Source
This commit is contained in:
hmo
2026-07-28 13:58:02 +08:00
parent 1dcadcb909
commit a3ef3b570f
2 changed files with 125 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/profile-scripts/technical_analysis.py
+72
View File
@@ -598,6 +598,77 @@ def analyze_volume_deep(code):
}
def _calc_scores(code: str) -> dict:
"""严谨技术指标：RSI/MACD/量能/背离——从stock_daily确定性计算"""
try:
import sqlite3
conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
rows = conn.execute(
"SELECT close, high, low, volume FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 30",
(code,)).fetchall()
conn.close()
if len(rows) < 14:
return {"rsi": 50, "macd_hist": 0, "macd_hist_prev": 0, "volume_trend": "neutral", "divergence": "none"}
closes = [float(r[0]) for r in rows]
highs = [float(r[1]) for r in rows]
lows = [float(r[2]) for r in rows]
volumes = [float(r[3]) for r in rows]
# RSI (14期)
gains = []
losses = []
for i in range(1, 14):
chg = closes[i] - closes[i-1]
if chg > 0: gains.append(chg)
else: losses.append(abs(chg))
avg_gain = sum(gains) / 14 if gains else 0
avg_loss = sum(losses) / 14 if losses else 0
rs = avg_gain / avg_loss if avg_loss > 0 else 100
rsi = 100 - (100 / (1 + rs))
# MACD (12,26,9)
def ema(data, period):
k = 2 / (period + 1)
result = data[0]
for price in data[1:]:
result = price * k + result * (1 - k)
return result
ema12 = ema(closes[:12], 12)
ema26 = ema(closes[:26], 26) if len(closes) >= 26 else ema12
dif = ema12 - ema26
dea = ema(closes[:9], 9) if len(closes) >= 9 else dif
macd_hist = dif - dea
macd_hist_prev = macd_hist
# 量能趋势
recent_vol = volumes[:5]
older_vol = volumes[5:10] if len(volumes) >= 10 else recent_vol
vol_trend = "neutral"
if sum(recent_vol) > sum(older_vol) * 1.2:
vol_trend = "accumulation"
elif sum(recent_vol) < sum(older_vol) * 0.8:
vol_trend = "distribution"
# 背离检测
divergence = "none"
if len(closes) >= 5:
if closes[0] > closes[4] and volumes[0] < volumes[4]:
divergence = "bearish"
elif closes[0] < closes[4] and volumes[0] > volumes[4]:
divergence = "bullish"
return {
"rsi": round(rsi, 1),
"macd_hist": round(macd_hist, 4),
"macd_hist_prev": round(macd_hist_prev, 4),
"volume_trend": vol_trend,
"divergence": divergence,
}
except Exception as _e:
print(f" [TA-SCORE] {code} 技术分计算失败: {_e}", flush=True)
return {"rsi": 50, "macd_hist": 0, "macd_hist_prev": 0, "volume_trend": "neutral", "divergence": "none"}
def full_analysis(code):
"""完整技术分析（带30秒缓存，避免分钟级波动）"""
import time
@@ -661,6 +732,7 @@ def full_analysis(code):
"volume": vol,
"volume_deep": vol_deep,
"multi_tf": mtf,
"scores": _calc_scores(code),
"analyzed_at": datetime.now().strftime("%H:%M"),
}
mofin_db.py
+53 -12
View File
@@ -1255,7 +1255,9 @@ def recompute_rr(conn, code: str) -> float:
# 基准参考价 = 区间中值（决定上方目标，三值共用）
ref = (el + eh) / 2.0 if el > 0 and eh > el else 0
target = tp
if ref > 0 and high_20d > ref and high_20d < tp:
# 已持仓股不适用20日新高阻力（用户已按原始推荐买入，RR应保持原值）
_owned = conn.execute("SELECT shares FROM holdings WHERE code=? AND is_active=1 AND shares>0", (code,)).fetchone()
if not _owned and ref > 0 and high_20d > ref and high_20d < tp:
target = high_20d
def _rr(x):
@@ -1282,18 +1284,32 @@ def recompute_rr(conn, code: str) -> float:
(rr_mid, rr_low, rr_high, code))
conn.commit()
compute_rec_score(conn, code) # RR 变→评分同步刷新
# RR<2.0 且信号为买入 → 降级为关注（2026-07-28 老爸：RR<2 的买入不推荐）
if rr_mid < 2.0:
_sig_r = conn.execute("SELECT timing_signal FROM holding_strategies WHERE code=? AND status=?", (code, "active")).fetchone()
if _sig_r and _sig_r[0] in ("买入", "可买入", "可加仓"):
conn.execute("UPDATE holding_strategies SET timing_signal=? WHERE code=? AND status=?", ("关注", code, "active"))
print(f" [RR-DOWNGRADE] {code} RR={rr_mid:.2f}<2.0, 买入→关注", flush=True)
return rr_mid
except Exception as e:
print(f" [RR] {code} 重算失败: {e}", flush=True)
return 0.0
def _calc_ta_score(code: str) -> dict:
"""严谨技术指标计算：RSI/MACD/量能/背离——从stock_daily确定性计算。
返回 {'rsi': float, 'macd_hist': float, 'macd_hist_prev': float, 'volume_trend': str, 'divergence': str}"""
try:
import technical_analysis as _ta_mod
_ta_full = _ta_mod.full_analysis(code)
if _ta_full and "error" not in _ta_full:
_scores = _ta_full.get("scores", {})
return {
'rsi': _scores.get("rsi", 50),
'macd_hist': _scores.get("macd_hist", 0),
'macd_hist_prev': _scores.get("macd_hist_prev", 0),
'volume_trend': _scores.get("volume_trend", "neutral"),
'divergence': _scores.get("divergence", "none"),
}
except Exception as _e:
print(f" [TA-SCORE] {code} 技术分计算失败: {_e}", flush=True)
return {'rsi': 50, 'macd_hist': 0, 'macd_hist_prev': 0, 'volume_trend': "neutral", 'divergence': "none"}
def compute_rec_score(conn, code: str) -> int:
"""五维复合推荐评分 0-100。RR高≠值得买，趋势+行业+信号综合判断。
维度：RR(0-35) + 信号(0-25) + 趋势(0-20) + 行业(0-10) + 区间(0-10)"""
@@ -1358,6 +1374,21 @@ def compute_rec_score(conn, code: str) -> int:
elif zone_pct >= 2: s_zone = 4
else: s_zone = 2
# ── 5. 严谨技术指标（2026-07-28 老爸：不许伪造好看的RR）──
_ta_s = _calc_ta_score(code)
_rsi = _ta_s["rsi"]
if 30 <= _rsi <= 70: s_trend += 3
elif _rsi < 30: s_trend += 5
elif _rsi > 70: s_trend -= 5
_mh, _mp = _ta_s["macd_hist"], _ta_s["macd_hist_prev"]
if _mh > 0 and _mh > _mp: s_trend += 5
elif _mh < 0 and _mh < _mp: s_trend -= 5
_vt = _ta_s["volume_trend"]
if _vt == "accumulation": s_trend += 5
elif _vt == "distribution": s_trend -= 5
_div = _ta_s["divergence"]
if _div == "bearish": s_trend -= 8
elif _div == "bullish": s_trend += 5
total = s_rr + s_sig + s_trend + s_sec + s_zone
conn.execute(
"UPDATE holding_strategies SET rec_score=? WHERE code=? AND status='active'",
@@ -1443,8 +1474,17 @@ def sync_recommend_tag(conn, code: str, timing_signal: str):
if timing_signal in _ACTION_BUY or timing_signal in _ACTION_SELL:
# ── 买入RR质量门禁（2026-07-27 老爸：RR<2.0的买入不值得推荐，tag都不能打，防盯盘垃圾）──
# 此前tag先打、enqueue再查RR，结果RR<2.0的tag已落盯盘，与XMPP不一致。
# 已持仓股不再重复推荐（2026-07-28 老爸：已买了的票该在持仓不在推荐）
_should_tag = True
if timing_signal in _ACTION_BUY:
_owned = conn.execute(
"SELECT shares FROM holdings WHERE code=? AND is_active=1 AND shares>0",
(code,)).fetchone()
if _owned and _owned[0] and _owned[0] > 0:
_should_tag = False # 已持仓，不再重复推荐
if _old_tag == 'current_recommend':
conn.execute("UPDATE holding_strategies SET tag='' WHERE code=? AND status='active'", (code,))
conn.commit()
elif timing_signal in _ACTION_BUY:
_rr_chk = conn.execute(
"SELECT rr_ratio FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
if _rr_chk and (_rr_chk[0] or 0) < 2.0:
@@ -1545,8 +1585,8 @@ def enqueue_recommend(conn, code: str):
def track_strategy_version(conn, code: str):
"""版本化策略追踪：每次策略变更(sync_recommend_tag 后)自动记录新版本。
只在 tag='current_recommend' 时记录；与上一条相比有实质变更才追加新版本。"""
"""版本化策略追踪：每次策略变更自动记录新版本。
跟踪所有策略状态（不限于 tag='current_recommend'），tag 清除时也记录。"""
try:
row = conn.execute(
"SELECT name, timing_signal, rec_score, rr_ratio, entry_low, entry_high, "
@@ -1555,8 +1595,9 @@ def track_strategy_version(conn, code: str):
if not row:
return
name, sig, score, rr, el, eh, sl, tp, pos, tag = row
if tag != 'current_recommend':
return # 非推荐状态不追踪
# 空壳策略（无信号/无买入区）不追踪
if not sig or (not el and not eh):
return
lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
price = lp[0] if lp and lp[0] else 0