# -*- coding: utf-8 -*- """猎隼本机盯盘循环（0额度·新浪免费接口） 每 N 秒拉取观察库 + 当日重点候选 + 全市场涨幅榜头部（候选发现）， 计算开幅/涨幅/反弹/成交额环比，检测异动（反弹加速/放量/翻红等）， 结果实时落盘 watch_trace_<日期>.jsonl + watch_snapshot_<日期>.json， 供主控（豆包）随时读取汇报，不依赖主控逐轮触发。 用法： python watch_loop.py --loop 90 # 每90秒一轮（默认） python watch_loop.py --once # 单轮 """ import json, os, sys, io, time, urllib.request, argparse, threading from datetime import datetime sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") HERE = os.path.dirname(os.path.abspath(__file__)) HDRS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0", "Referer": "https://finance.sina.com.cn/"} def now(): return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def hq(codes): """批量拉实时行情，返回 {code: dict}。字段：name/open/pre/price/high/low/amt/vol/date/time""" if not codes: return {} url = "https://hq.sinajs.cn/list=" + ",".join(("sh" if c.startswith("6") else "sz") + c for c in codes) out = {} try: req = urllib.request.Request(url, headers=HDRS) with urllib.request.urlopen(req, timeout=15) as r: raw = r.read().decode("gbk") except Exception as e: print("[%s] hq fail: %s" % (now(), e)) return {} for ln in raw.strip().split("\n"): if "=" not in ln: continue head = ln.split("=")[0] code = head[-6:] p = ln.split("=", 1)[1].strip().strip('";').split(",") if len(p) < 32: continue try: out[code] = { "name": p[0], "open": float(p[1]), "pre": float(p[2]), "price": float(p[3]), "high": float(p[4]), "low": float(p[5]), "amt": float(p[9]) / 1e8, "vol": float(p[8]), "date": p[30], "time": p[31], } except (TypeError, ValueError, IndexError): continue return out def idx_quote(): """拉上证指数实时涨跌（sh000001），返回 {chg, price} 或 None。用于相对强度计算 师傅10-08定稿方向：大盘跌它不跌反涨=当日最强票；大盘稍拉它马上跑=进攻性强""" try: req = urllib.request.Request("https://hq.sinajs.cn/list=s_sh000001", headers=HDRS) with urllib.request.urlopen(req, timeout=15) as r: raw = r.read().decode("gbk") if "=" not in raw: return None p = raw.split("=", 1)[1].strip().strip('";').split(",") if len(p) < 4: return None # s_sh000001 格式: 名称,当前点,涨跌额,涨跌幅%,成交量,成交额,... price = float(p[1]) chg = float(p[3].replace("%", "")) return {"price": price, "chg": chg} except Exception as e: print("[%s] 指数拉取失败: %s" % (now(), e)) return None def top_gainers(topn=120): """新浪分页接口拉全市场涨幅榜（拉8页覆盖涨幅1%+刚启动区，师傅10-08定稿： 只看头部=全是涨6-10%的，1-3%刚启动票被抹杀，必须扫到低涨幅区）""" base = "https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/Market_Center.getHQNodeData" rows = [] for page in range(1, 9): url = "%s?page=%d&num=%d&sort=changepercent&asc=0&node=hs_a" % (base, page, 100) try: req = urllib.request.Request(url, headers=HDRS) with urllib.request.urlopen(req, timeout=15) as r: arr = json.loads(r.read().decode("utf-8")) except Exception as e: print("[%s] top fail page %d: %s" % (now(), page, e)) break if not arr: break page_min = min((float(x.get("changepercent") or 0) for x in arr), default=0) for x in arr: code = x.get("code", "") if not (code.startswith("60") or code.startswith("00")): continue name = x.get("name", "") if "ST" in name.upper() or "退" in name or name[:1] == "N": continue rows.append(code) time.sleep(0.1) # 整页最大涨幅 < 0.5% 时停（已覆盖刚启动区，再往下是绿盘区） if page_min < 0.5: break return list(dict.fromkeys(rows)) def top_amount(topn=300): """成交额榜：资金先进场（成交额大）但价格未必涨=启动早期票的发现源（师傅10-08定稿核心）""" base = "https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/Market_Center.getHQNodeData" rows = [] for page in range(1, 4): url = "%s?page=%d&num=%d&sort=amount&asc=0&node=hs_a" % (base, page, 100) try: req = urllib.request.Request(url, headers=HDRS) with urllib.request.urlopen(req, timeout=15) as r: arr = json.loads(r.read().decode("utf-8")) except Exception as e: print("[%s] amount fail page %d: %s" % (now(), page, e)) break if not arr: break for x in arr: code = x.get("code", "") if not (code.startswith("60") or code.startswith("00")): continue name = x.get("name", "") if "ST" in name.upper() or "退" in name or name[:1] == "N": continue rows.append(code) time.sleep(0.1) return list(dict.fromkeys(rows)) def load_vol5(): """从缓存K线近似5日均量（量比分母），返回 {code: vol5}。缓存大只加载一次""" cache_file = os.path.join(HERE, "backtest_kline_cache.json") idx = {} try: with open(cache_file, "r", encoding="utf-8") as f: data = json.load(f) except Exception as e: print("缓存读取失败(量比标暂无): %s" % e) return idx for code, v in data.items(): kl = [r for r in v.get("klines", []) if len(r) == 6] if len(kl) < 5: continue idx[code] = sum(float(r[5]) for r in kl[-5:]) / 5.0 return idx def elapsed_ratio(ts): """盘中已过交易时段比例（9:30-11:30 / 13:00-15:00），用于量比折算""" hm = ts[11:16] if "09:3" <= hm < "10:0": return 0.02 if hm < "11:30": return (int(hm[0:2]) * 60 + int(hm[3:5]) - 570) / 240.0 if hm < "13:00": return 0.5 if hm < "15:00": return 0.5 + (int(hm[0:2]) * 60 + int(hm[3:5]) - 780) / 240.0 return 1.0 def is_trading_time(ts): """交易时段判断（师傅2026-10-10拍板：休市沿用最近有效行情） 交易日 9:30-11:30 / 13:00-15:00 为交易时段；周末全天休市； 节假日/非交易时段（早于9:30、午休、收盘后）一律按休市处理。 休市时不做行情拉取与 judge 重算，避免量比失真导致假"可做"堆积。 """ try: # 周末（周6=5、周日=6）直接休市 wd = datetime.strptime(ts[:10], "%Y-%m-%d").weekday() if wd >= 5: return False hm = ts[11:16] if "09:30" <= hm < "11:30": return True if "13:00" <= hm < "15:00": return True return False except Exception: return False def load_watch(): try: with open(os.path.join(HERE, "watchlist.json"), "r", encoding="utf-8") as f: data = json.load(f) return [s["code"] for s in data.get("stocks", [])] except Exception as e: print("watchlist 读取失败: %s" % e) return [] def calc_judge(r): """可做/观察/不可做判断（师傅2026-10-09拍板：20秒重算一次，纯规则零额度） 按通道方向分级判断（师傅再定稿）： 强势上涨->可做（重心上移最强）；上涨通道->用量比+强度判断 下跌通道->观察（可能再爬）；下跌警告->不可做（跌破前低） 原票/无通道->走原逻辑（量比>=2.5 + 相对强度>=1.5）""" chg = r["chg"]; rb = r["rebound"]; stage = r["stage"]; tag = r["tag"] vr = r["vol_ratio"]; rel = r["rel"] channel = r.get("channel", "") or "" vr_f = vr if isinstance(vr, (int, float)) else None rel_f = rel if isinstance(rel, (int, float)) else None # 量比统一带进理由（师傅2026-10-09再定稿：理由里要看到量比多少） vr_s = ("%.1f" % vr_f) if isinstance(vr_f, (int, float)) else "暂无" # 不可做优先（所有票通用）：涨幅超6%=追高尾声区；量比<1=无量 if chg > 6.0: return ("不做", "涨幅%.2f%%超6%%追高区·量比%s" % (chg, vr_s)) if vr_f is not None and vr_f < 1.0: return ("不做", "无量·资金未进·量比%s" % vr_s) # 通道方向分级（师傅2026-10-09定稿） if channel == "强势上涨": return ("可做", "强势上涨·重心上移·量比%s" % vr_s) if channel == "下跌警告": return ("不做", "下跌警告·跌破前低·量比%s" % vr_s) if channel == "下跌通道": # 观察：可能再次往上爬，暂不做 return ("观察", "下跌通道·等止跌企稳·量比%s" % vr_s) # 强势上涨已直接可做；上涨通道/原票/无通道走量比+强度判断 ok_vr = vr_f is not None and vr_f >= 2.5 ok_rel = (rel_f is not None and rel_f >= 1.5) or tag == "逆势抗跌" if 1.0 <= chg <= 6.0 and ok_vr and ok_rel: reason = stage if tag: reason += "·%s" % tag reason += "·量比%s" % vr_s return ("可做", reason) # 观察：涨幅在区但量能/强度不足，等量能确认 reason = "%s·%s·量比%s" % (stage, tag if tag else "相对强度不足", vr_s) return ("观察", reason) def main(): ap = argparse.ArgumentParser() ap.add_argument("--loop", type=int, default=90, help="循环间隔秒") ap.add_argument("--once", action="store_true") args = ap.parse_args() watch = load_watch() vol5 = load_vol5() last_judge = [0.0] # 可做判断20秒重算一次（师傅2026-10-09拍板·按猎隼口径） judge_map = {} # {code: (judge, reason)} chan_map = {} # {code: {'hi':历史最高涨幅%, 'lo':历史最低涨幅%}} 上涨/下跌通道跟踪（师傅10-09定稿） seen_codes = set() # 持续跟踪池：见过的票持续盯，不掉出榜单就丢（师傅10-09定稿） print("[%s] 观察库 %d 只 | 量比缓存 %d 只 | 本机盯盘启动" % (now(), len(watch), len(vol5))) def run_round(): ts = now() date = ts[:10] # 【休市沿用最近有效行情·师傅2026-10-10拍板】 # 周末/非交易时段不做行情拉取与 judge 重算（休市量比失真→假"可做"堆积）， # 直接把最近一次有效快照复制覆盖当日文件，看板读到的是最近交易日有效行情，可作参考。 if not is_trading_time(ts): snap_dir = HERE # 找最近一个已落盘的有效快照（非当日的最近文件；当日也可能已由早盘生成） last_valid = None try: files = [f for f in os.listdir(snap_dir) if f.startswith("watch_snapshot_") and f.endswith(".json")] files.sort() for f in reversed(files): d = f[len("watch_snapshot_"):-len(".json")] if d < date: # 只取严格早于今天的有效快照（最近交易日） last_valid = f break except Exception: last_valid = None if last_valid: try: import shutil shutil.copy(os.path.join(snap_dir, last_valid), os.path.join(snap_dir, "watch_snapshot_%s.json" % date)) print("[%s] 休市沿用最近有效行情快照 %s（不重算judge，避免休市假可做）" % (ts, last_valid)) except Exception as e: print("[%s] 休市沿用快照失败: %s" % (ts, e)) else: print("[%s] 休市且无历史有效快照，跳过本轮" % ts) return codes = list(watch) # 发现源双榜：涨幅榜头部（已异动）+ 成交额榜（资金先进=启动早期源） new_codes = top_gainers(120) + top_amount(300) for c in new_codes: if c not in codes: codes.append(c) # 持续跟踪池：把跟踪池里记住的票也带上（限300只，防止无限膨胀；进过视野就持续盯） for c in seen_codes: if c not in codes: codes.append(c) if len(codes) > 800: codes = codes[:800] q = hq(codes) # 每轮把见过的票记入持续跟踪池（只在主板60/00范围内） for c in q: if (c.startswith("60") or c.startswith("00")) and len(seen_codes) < 500: seen_codes.add(c) # 重点候选【师傅定稿·启动早期票】：只炒"涨的过程中能抓到"的低开慢拉票 # 开幅 -1~-5（低开优先，微低开-1~0也收） 且 当日涨幅 1%~8%（师傅10-08补充：反弹7-8%也能吃五六个点七八个点） # 量比>=1.2（放量确认，资金真进）+ 成交额>=0.5亿 # 已涨停（>8%接近涨停/涨停）不入选——只记L3次日竞价，不当日炒 # 买点阶段（师傅10-08再定稿）：刚启动=反弹1~2%（最优，涨幅刚起量能刚放大）> 黄金买点=反弹2~3.5% > 中段3.5~6% > 尾声6~8% focus = [] er = elapsed_ratio(ts) ix = idx_quote() idx_chg = ix["chg"] if ix else None for code, d in q.items(): # 板块过滤（对齐猎隼10-09指令：只做主板60/00，排除创业板30/科创板68/北交所/ST/退市/N新股） if not (code.startswith("60") or code.startswith("00")): continue nm = d.get("name", "") if "ST" in nm.upper() or "退" in nm or nm[:1] == "N": continue oc = (d["open"] - d["pre"]) / d["pre"] * 100 if d["pre"] > 0 else 0 chg = (d["price"] - d["pre"]) / d["pre"] * 100 if d["pre"] > 0 else 0 rb = chg - oc v5 = vol5.get(code) # 量比=当日累计量/(5日均量×已过时段比例)——盘中折算，防低估误杀 vr = (d["vol"] / (v5 * er)) if (v5 and v5 > 0 and er > 0) else None # 【通道跟踪：连续涨跌计数】（师傅2026-10-09定稿） # chan_map[code] = {'prev':上一轮涨幅, 'up':连续上涨次数, 'down':连续下跌次数, 'hi':前高, 'lo':前低, 'chan':标签} m = chan_map.get(code) if m is None: m = {"prev": chg, "up": 0, "down": 0, "hi": chg, "lo": chg, "chan": "观察"} else: # 连续涨跌计数：本轮相对上轮涨幅 delta = chg - m["prev"] if delta >= 0.8: m["up"] += 1; m["down"] = 0 elif delta <= -0.8: m["down"] += 1; m["up"] = 0 else: m["up"] = 0; m["down"] = 0 # 前高/前低更新 if chg > m["hi"]: m["hi"] = chg if chg < m["lo"]: m["lo"] = chg m["prev"] = chg chan_map[code] = m # 判定（师傅10-09定稿）：连续3次上涨(每次>=0.8%) = 上涨通道；涨过前高 = 强势上涨 # 连续3次下跌(每次>=0.8%) = 下跌通道；跌破前低 = 下跌警告 if m["up"] >= 3 and chg >= m["hi"]: chandir = "强势上涨" # 连续3涨且破前高 elif m["up"] >= 3: chandir = "上涨通道" # 连续3涨 elif m["down"] >= 3 and chg <= m["lo"]: chandir = "下跌警告" # 连续3跌且破前低 elif m["down"] >= 3: chandir = "下跌通道" # 连续3跌 else: chandir = "观察" # 原票（师傅定稿硬条件）：低开-6~0 + 涨幅1-8 + 反弹>0.5 + 成交额>=0.5 # 量比不做硬门槛（10-08节后首日缩量，硬过滤=误杀），只作标注+排序 is_orig = (-6.0 <= oc <= 0.0 and 1.0 <= chg <= 8.0 and rb > 0.5 and d["amt"] >= 0.5) # 通道票（新增捕捉源）：不满足原硬条件，但有通道标签(涨/跌方向)且可做基本条件 is_chan = (not is_orig) and chandir in ("强势上涨", "上涨通道", "下跌通道", "下跌警告") and 1.0 <= chg <= 8.0 and d["amt"] >= 0.5 if is_orig or is_chan: # 买点阶段四级（师傅2026-10-09定稿）：反弹1-2%=最优买点；2-3.5%=黄金买点；3.5-6%=标准买点；6-8%=追高警戒 if 1.0 <= rb <= 2.0: stage = "最优买点" elif 2.0 < rb <= 3.5: stage = "黄金买点" elif 3.5 < rb <= 6.0: stage = "标准买点" else: stage = "追高警戒" # 相对强度=个股涨幅-大盘涨幅（师傅10-08定稿：大盘跌它不跌=最强；大盘拉它先跑=进攻） rel = (chg - idx_chg) if idx_chg is not None else None # 抗跌/进攻标注：大盘下跌时个股红盘=逆势抗跌；大盘上涨时个股相对强度大=进攻强 tag = "" if rel is not None: if idx_chg < -0.3 and chg > 0 and rel >= 1.5: tag = "逆势抗跌" elif idx_chg >= 0 and rel >= 2.0: tag = "进攻强" # 原票不打通道标签；通道票标通道方向（名称后显示：上涨通道/强势上涨/下跌通道/下跌警告） chan_label = chandir if is_chan else "" focus.append({"ts": ts, "code": code, "name": d["name"], "open": round(d["open"], 2), "price": round(d["price"], 2), "open_chg": round(oc, 2), "chg": round(chg, 2), "rebound": round(rb, 2), "amt": round(d["amt"], 2), "vol_ratio": round(vr, 2) if vr else "暂无", "stage": stage, "idx_chg": round(idx_chg, 2) if idx_chg is not None else "暂无", "rel": round(rel, 2) if rel is not None else "暂无", "tag": tag, "channel": chandir, "is_chan": is_chan, "chan_label": chan_label}) # 刚启动/黄金买点优先，同阶段内量比大优先（资金真进），量比暂无放最后 stage_rank = {"最优买点": 0, "黄金买点": 1, "标准买点": 2, "追高警戒": 3} focus.sort(key=lambda r: (stage_rank.get(r["stage"], 9), -(r["vol_ratio"] if isinstance(r["vol_ratio"], float) else -1), -r["amt"])) # 可做判断（20秒重算一次；行情30秒轮） now_epoch = time.time() if not judge_map or now_epoch - last_judge[0] >= 20: for r in focus: judge_map[r["code"]] = calc_judge(r) last_judge[0] = now_epoch for r in focus: j, rs = judge_map.get(r["code"], ("暂无", "暂无")) r["judge"] = j r["reason"] = rs # 落盘轨迹 trace_file = os.path.join(HERE, "watch_trace_%s.jsonl" % date) with open(trace_file, "a", encoding="utf-8") as f: for r in focus: f.write(json.dumps(r, ensure_ascii=False) + "\n") # 快照 snap = {"ts": ts, "date": date, "focus": focus[:40], "total": len(focus)} snap_file = os.path.join(HERE, "watch_snapshot_%s.json" % date) with open(snap_file, "w", encoding="utf-8") as f: json.dump(snap, f, ensure_ascii=False, indent=1) # 控制台摘要（只打前10，省输出） print("[%s] 实时%4d只 | 大盘%+5.2f%% | 启动早期候选%3d只 | 前10:" % (ts, len(q), idx_chg if idx_chg is not None else 0, len(focus))) for r in focus[:10]: print(" %s %-8s 现%7.2f 开幅%+5.2f 涨幅%+5.2f 反弹%+5.2f 量比%6s 额%5.2f亿 %s %s %s" % ( r["code"], r["name"], r["price"], r["open_chg"], r["chg"], r["rebound"], str(r["vol_ratio"]), r["amt"], r["stage"], r["tag"], r["rel"])) if args.once: run_round() print("单轮完成") else: while True: try: run_round() except Exception as e: print("[%s] 轮次异常: %s" % (now(), e)) time.sleep(args.loop) if __name__ == "__main__": main()