#!/usr/bin/env python3 """300308 午后紧盯监控 (no_agent) 每2分钟运行一次。在条件满足时输出买入信号，否则静默。 支持策略动态调整——输出状态变化。 """ import json, os, sys, urllib.request from pathlib import Path from datetime import datetime STATE_PATH = "/home/hmo/.hermes/300308_monitor_state.json" BUY_ZONE_LOW = 1307.22 BUY_ZONE_HIGH = 1358.94 STOP_LOSS = 1293.88 TAKE_PROFIT = 1456.53 # 三档触发条件（按优先级） LEVEL_STRONG = {"label": "★ 强信号", "price_min": 1330, "price_breach": 1325, "desc": "前3根5分K站稳1330+量放"} LEVEL_MID = {"label": "◎ 中信号", "price_min": 1315, "price_breach": 1307, "desc": "回踩1315企稳+买盘放量"} LEVEL_WEAK = {"label": "○ 弱信号", "price_min": 1307, "price_breach": 1298, "desc": "在买入区下沿附近企稳"} def get_price(): url = "https://qt.gtimg.cn/q=sz300308" req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) resp = urllib.request.urlopen(req, timeout=10) data = resp.read().decode("gbk") parts = data.split("~") price = float(parts[3]) high = float(parts[33]) low = float(parts[34]) change_pct = float(parts[32]) volume = int(parts[6]) buy_vol = int(parts[7]) if parts[7] else 0 sell_vol = int(parts[8]) if parts[8] else 0 return { "price": price, "high": high, "low": low, "change_pct": change_pct, "volume": volume, "buy_vol": buy_vol, "sell_vol": sell_vol, } def load_state(): try: with open(STATE_PATH) as f: return json.load(f) except: return {"phase": "waiting", "triggered_levels": [], "last_report": None, "afternoon_low": 99999} def save_state(s): os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True) with open(STATE_PATH, "w") as f: json.dump(s, f, ensure_ascii=False, indent=2) def main(): now = datetime.now() # 只在13:00-15:00运行 if now.hour < 13 or now.hour >= 15: # After 15:00, send final summary if not already sent state = load_state() if state.get("phase") not in ("done", "final"): # Market closed, output final status state["phase"] = "final" save_state(state) print("【300308紧盯结束】15:00收盘。") # Final check price = get_price() if price["price"] >= STOP_LOSS: print(f"收盘价{price['price']:.2f}，未触及止损{STOP_LOSS}，策略有效。") else: print(f"⚠️ 收盘价{price['price']:.2f}已跌破止损{STOP_LOSS}！") return state = load_state() # If already in done phase (signal already sent), stay silent if state.get("phase") == "done": return price_data = get_price() p = price_data["price"] vol = price_data["volume"] buy_vol = price_data["buy_vol"] sell_vol = price_data["sell_vol"] # Track afternoon low if p < state["afternoon_low"]: state["afternoon_low"] = p # Get volume since last check to calculate relative activity state.setdefault("last_vol", 0) vol_delta = vol - state["last_vol"] state["last_vol"] = vol vol_active = vol_delta > 0 # new trades happened # Determine buy/sell pressure (active buying as proportion) total_trade = buy_vol + sell_vol buy_ratio = buy_vol / total_trade if total_trade > 0 else 0.5 # Minutes into afternoon session afternoon_minutes = (now.hour - 13) * 60 + now.minute candle_num = afternoon_minutes // 5 + 1 # Which 5-min candle we're in # ---------------------------------------------------------------- # Level checking - try from strong to weak # ---------------------------------------------------------------- triggered = state.get("triggered_levels", []) # Check levels from weak to strong for level_key, level in [("weak", LEVEL_WEAK), ("mid", LEVEL_MID), ("strong", LEVEL_STRONG)]: if level_key in triggered: continue # already triggered, skip if p >= level["price_min"] and state["afternoon_low"] >= level["price_breach"]: # Basic price conditions met — check volume # Volume should show active buying (buy_vol > sell_vol * 0.8 = not too one-sided) volume_ok = buy_ratio >= 0.45 # at least 45% buying if volume_ok: triggered.append(level_key) state["triggered_levels"] = triggered if level_key == "strong": msg = ( f"【300308买入信号】★ 强信号触发\n" f"现价{p:.2f}，已在1330上方站稳，午后最低{state['afternoon_low']:.2f}未破1325\n" f"主动买比{buy_ratio:.0%}，买盘积极\n" f"距止损{(p-STOP_LOSS)/STOP_LOSS*100:.1f}%，距止盈{(TAKE_PROFIT-p)/p*100:.1f}%\n" f"操作：买入1手(100股)，止损{STOP_LOSS}，止盈{TAKE_PROFIT}" ) state["phase"] = "done" save_state(state) print(msg) return if level_key == "mid": msg = ( f"【300308买入信号】◎ 中信号触发\n" f"现价{p:.2f}，回踩1315企稳，午后最低{state['afternoon_low']:.2f}\n" f"主动买比{buy_ratio:.0%}，买盘回温\n" f"距止损{(p-STOP_LOSS)/STOP_LOSS*100:.1f}%，距止盈{(TAKE_PROFIT-p)/p*100:.1f}%\n" f"操作：可买入1手(100股)，止损{STOP_LOSS}(-{(p-STOP_LOSS)/p*100:.1f}%)，止盈{TAKE_PROFIT}" ) state["phase"] = "done" save_state(state) print(msg) return # ---------------------------------------------------------------- # Check for failure conditions # ---------------------------------------------------------------- if state["afternoon_low"] < 1300: # Breaking towards stop loss if not state.get("warned_low"): state["warned_low"] = True save_state(state) print(f"【300308风险警告】午后最低{state['afternoon_low']:.2f}，逼近止损{STOP_LOSS}(-4.4%)。" f"现价{p:.2f}。建议取消今日买入。") return if state["afternoon_low"] < STOP_LOSS: state["phase"] = "done" # stop monitoring save_state(state) print(f"【300308止损触发】午后最低{state['afternoon_low']:.2f}<止损{STOP_LOSS}。" f"策略已失效，取消今日买入。等明日重新评估。") return # ---------------------------------------------------------------- # Periodically report status (every ~10 mins) # ---------------------------------------------------------------- last_report = state.get("last_report_min", -999) report_interval = 10 # minutes this_minute = now.hour * 60 + now.minute if this_minute - last_report >= report_interval: state["last_report_min"] = this_minute # Check what condition we're closest to if p >= LEVEL_STRONG["price_min"]: closest = "强信号(1330)" elif p >= LEVEL_MID["price_min"]: closest = f"中信号(1315)，差{LEVEL_STRONG['price_min']-p:.1f}到强信号" else: closest = f"弱信号(1307)，差{LEVEL_MID['price_min']-p:.1f}到中信号" msg = ( f"【300308午后监控】第{candle_num}根5分K线\n" f"现价{p:.2f} | 午后最低{state['afternoon_low']:.2f}\n" f"买比{buy_ratio:.0%} | 距目标{closest}\n" f"仍有效，继续监控。" ) save_state(state) print(msg) return save_state(state) # Silent - nothing to report if __name__ == "__main__": main()