#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ XAU-USDT-SWAP 自动盯盘脚本 - 每 3 分钟自动刷新行情, 计算价格/成交量/均线(SMA/EMA)/RSI/MACD - 判断趋势(上涨/下跌/震荡), 给出 3~15 分钟操作建议(多/空/观望)+ 置信度 + 支撑阻力 - 自动运行(可后台循环), 日志写入 xau_monitor.log, 行情突变(价/量异动)本地提醒 - 循环间隔内置刷新倒计时显示(格式 分:秒), 归零自动触发刷新并重新计时 - 读取账户当前 XAU 持仓/挂单(只读), 结合实时行情给出持仓操作建议(含保本最低收益率/保本价) - 实时抓取金十 7x24 新闻, 按标的自动关联, 给出≤10字核心说明与利多/利空倾向; 盈亏率着色(盈红亏绿) - 支持 --inst 指定盯盘标的(如 BTC-USDT-SWAP / ETH-USDT-SWAP), 新闻与建议随标的自动适配 用法: python -X utf8 xau_monitor.py --dns-ip 198.18.0.0 # 前台循环, 每3分钟一次 python -X utf8 xau_monitor.py --dns-ip 198.18.0.0 --once # 仅跑一次 python -X utf8 xau_monitor.py --dns-ip 198.18.0.0 --alert-popup # 突变时弹窗(需 Windows) """ import json, hmac, hashlib, base64, datetime, sys, os, argparse, subprocess, time, math, re HERE = os.path.dirname(os.path.abspath(__file__)) KEYS_FILE = os.path.join(HERE, "okx_keys.json") LOG_FILE = os.path.join(HERE, "xau_monitor.log") INST = "XAU-USDT-SWAP" INTERVAL = 180 # 3 分钟 # ---------- OKX 请求(走 fake-ip 绕过 DNS 污染 + curl 绕过 WAF) ---------- def load_keys(): with open(KEYS_FILE, "r", encoding="utf-8") as f: return json.load(f) def sign(secret, ts, method, path, body=""): pre = ts + method.upper() + path + body return base64.b64encode(hmac.new(secret.encode(), pre.encode(), hashlib.sha256).digest()).decode() def okx_get(path, keys, dns_ip): ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" sig = sign(keys["secret"], ts, "GET", path, "") domain = "www.okx.com" cmd = ["curl", "-s", "--max-time", "15", "--resolve", f"{domain}:443:{dns_ip}"] cmd += ["-H", f"OK-ACCESS-KEY:{keys['api_key']}", "-H", f"OK-ACCESS-SIGN:{sig}", "-H", f"OK-ACCESS-TIMESTAMP:{ts}", "-H", f"OK-ACCESS-PASSPHRASE:{keys['passphrase']}", "-H", "Content-Type:application/json", "-H", "User-Agent:curl/8.5.0", f"https://{domain}{path}"] try: return json.loads(subprocess.run(cmd, capture_output=True, text=True, timeout=20).stdout) except Exception as e: return {"code": "-1", "msg": str(e)} # ---------- 终端着色(盈亏率: 盈利红 / 亏损绿) ---------- _ANSI_RED = "\033[31m" _ANSI_GREEN = "\033[32m" _ANSI_RESET = "\033[0m" def enable_color(): """Windows 10+ 启用虚拟终端以显示 ANSI 颜色; 非 Windows/不支持则静默忽略.""" if os.name == "nt": try: os.system("") # 触发启用 VT100 处理 except Exception: pass def red(s): return f"{_ANSI_RED}{s}{_ANSI_RESET}" def green(s): return f"{_ANSI_GREEN}{s}{_ANSI_RESET}" def strip_ansi(s): """去除 ANSI 转义, 用于写日志文件避免乱码.""" return re.sub(r"\033\[[0-9;]*m", "", s) # ---------- 实时新闻(金十 7x24, 走 fake-ip) ---------- def news_profile(inst): """根据标的代码返回新闻关键词配置. 返回 dict: keywords(相关性过滤词), bias_bull(利好), bias_bear(利空), asset(中文名). 宏观类关键词(美联储/美元/降息/加息/避险/地缘/非农/通胀/央行/美债) 对所有品种通用, 叠加品种专属词.""" macro = ["美联储", "美元", "降息", "加息", "避险", "地缘", "非农", "通胀", "央行", "美债", "杰克逊霍尔", "就业", "GDP", "PMI"] i = (inst or "").upper() if "XAU" in i or "黄金" in inst or "GOLD" in i: return dict(keywords=macro + ["黄金", "XAU", "白银", "贵金属", "购金", "增持黄金", "金矿"], bias_bull=["降息", "避险", "地缘", "宽松", "走弱", "下跌", "购金", "增持黄金"], bias_bear=["加息", "鹰派", "走强", "上涨", "紧缩", "强美元", "抛售", "减持黄金"], asset="黄金") if "BTC" in i or "比特币" in inst: return dict(keywords=macro + ["比特币", "BTC", "加密", "币圈", "ETF", "矿机", "灰度"], bias_bull=["降息", "宽松", "走弱", "下跌", "ETF", "流入", "减半", "采用"], bias_bear=["加息", "鹰派", "走强", "上涨", "紧缩", "强美元", "流出", "监管", "禁令"], asset="比特币") if "ETH" in i or "以太" in inst: return dict(keywords=macro + ["以太坊", "ETH", "加密", "币圈", "ETF", "灰度", "Layer2"], bias_bull=["降息", "宽松", "走弱", "下跌", "ETF", "流入", "升级", "采用"], bias_bear=["加息", "鹰派", "走强", "上涨", "紧缩", "强美元", "流出", "监管", "禁令"], asset="以太坊") # 默认(股票/其他): 仅宏观 + 标的自身代码作为关键词 return dict(keywords=macro + [inst], bias_bull=["降息", "宽松", "走弱", "下跌", "利好", "超预期", "回购", "增持"], bias_bear=["加息", "鹰派", "走强", "上涨", "紧缩", "利空", "不及预期", "减持"], asset=inst) def get_news(dns_ip, inst, limit=5): """抓取金十首页内嵌快讯, 过滤与标的(inst)相关的新闻, 返回 list[dict(title,brief,bias)]. brief 为核心说明(≤10字); bias: 'bull'利好 / 'bear'利空 / 'neutral'中性(均相对该标的).""" domain = "www.jin10.com" cmd = ["curl", "-s", "--max-time", "15", "--resolve", f"{domain}:443:{dns_ip}", "-H", "User-Agent:curl/8.5.0", "-H", "Referer:https://www.jin10.com/", f"https://{domain}/"] try: html = subprocess.run(cmd, capture_output=True, text=True, timeout=20).stdout except Exception: return [] if not html: return [] prof = news_profile(inst) NEWS_KEYWORDS = prof["keywords"] BIAS_BULL = prof["bias_bull"] BIAS_BEAR = prof["bias_bear"] # 提取快讯标题: 优先 flash-top-list__item-content 文本, 兜底 title="..." 属性 titles = re.findall(r'flash-top-list__item-content">(.*?)', html, re.S) if not titles: kw_alt = "|".join(NEWS_KEYWORDS[:12]) titles = re.findall(r'title="([^"]*(?:' + kw_alt + r')[^\"]*)"', html) # 清洗标签 cleaned = [] for t in titles: t = re.sub(r"<[^>]+>", "", t).strip() if t and len(t) > 4: cleaned.append(t) # 去重 + 过滤相关词 seen, rel = set(), [] for t in cleaned: if t in seen: continue seen.add(t) if any(k in t for k in NEWS_KEYWORDS): rel.append(t) if len(rel) >= limit * 3: break # 生成说明与倾向 out = [] for t in rel[:limit]: brief = t[:10] # 单条不超过10个字 low = t if any(b in low for b in BIAS_BULL) and not any(b in low for b in BIAS_BEAR): bias = "bull" elif any(b in low for b in BIAS_BEAR): bias = "bear" else: bias = "neutral" out.append({"title": t, "brief": brief, "bias": bias}) return out # ---------- 技术指标 ---------- def sma(vals, n): if len(vals) < n: return None return sum(vals[-n:]) / n def ema(vals, n): if len(vals) < n: return None k = 2 / (n + 1) e = vals[0] for v in vals[1:]: e = v * k + e * (1 - k) return e def rsi(closes, n=14): if len(closes) < n + 1: return None gains, losses = [], [] for i in range(1, len(closes)): d = closes[i] - closes[i - 1] gains.append(max(d, 0)); losses.append(max(-d, 0)) # Wilder smoothing ag = sum(gains[:n]) / n; al = sum(losses[:n]) / n for i in range(n, len(gains)): ag = (ag * (n - 1) + gains[i]) / n al = (al * (n - 1) + losses[i]) / n if al == 0: return 100.0 rs = ag / al return 100 - 100 / (1 + rs) def macd(closes, fast=12, slow=26, sig=9): if len(closes) < slow + sig: return None, None, None ema_f = ema(closes, fast) # 需逐根计算保证连续 kf, ks, kg = 2 / (fast + 1), 2 / (slow + 1), 2 / (sig + 1) ef, es = closes[0], closes[0] ef_arr, es_arr = [], [] for v in closes: ef = v * kf + ef * (1 - kf); es = v * ks + es * (1 - ks) ef_arr.append(ef); es_arr.append(es) dif = [ef_arr[i] - es_arr[i] for i in range(len(closes))] dea = dif[0] dea_arr = [dea] for v in dif[1:]: dea = v * kg + dea * (1 - kg); dea_arr.append(dea) dif_v, dea_v = dif[-1], dea_arr[-1] return dif_v, dea_v, (dif_v - dea_v) def sar(highs, lows, af_start=0.02, af_max=0.20): """抛物线转向 SAR (Wilder). 返回最近一根的 SAR 值, 及当前多/空方向. highs/lows 为升序(旧->新)序列.""" if len(highs) < 2: return None, None # 初始方向: 比较前两根收盘用高低判断 uptrend = highs[1] >= highs[0] if uptrend: sar = lows[0] ep = highs[1] else: sar = highs[0] ep = lows[1] af = af_start for i in range(2, len(highs)): prev_sar = sar if uptrend: sar = prev_sar + af * (ep - prev_sar) # 不能高于前两根低 sar = min(sar, lows[i - 1], lows[i - 2] if i >= 2 else lows[i - 1]) if highs[i] > ep: ep = highs[i]; af = min(af + af_start, af_max) if lows[i] < sar: # 反转 uptrend = False; sar = ep; ep = lows[i]; af = af_start else: sar = prev_sar + af * (ep - prev_sar) # 不能低于前两根高 sar = max(sar, highs[i - 1], highs[i - 2] if i >= 2 else highs[i - 1]) if lows[i] < ep: ep = lows[i]; af = min(af + af_start, af_max) if highs[i] > sar: # 反转 uptrend = True; sar = ep; ep = highs[i]; af = af_start return sar, ("多" if uptrend else "空") # ---------- 持仓/挂单查询(只读) + 操作建议 ---------- def get_positions_and_orders(keys, dns_ip, inst): """查询指定标的(inst)的当前持仓与未成交挂单(只读, 不下单). 返回 (positions, orders), 任一查询失败则该部分为空列表.""" positions, orders = [], [] try: p = okx_get(f"/api/v5/account/positions?instType=SWAP&instId={inst}", keys, dns_ip) if p.get("code") == "0": for it in p.get("data", []): if float(it.get("pos", 0) or 0) != 0: # 仅保留有实际仓位的 positions.append(it) except Exception: pass try: o = okx_get(f"/api/v5/trade/orders-pending?instType=SWAP&instId={inst}", keys, dns_ip) if o.get("code") == "0": for it in o.get("data", []): orders.append(it) except Exception: pass return positions, orders def position_advice(positions, orders, d, fee_rate=0.0005): """结合实时行情 d, 给出每个持仓/挂单的操作建议. 返回字符串列表. fee_rate: 单边 taker 手续费率(默认 0.05%), 保本按 开仓+平仓 两次计.""" lines = [] last = d["last"] sar_v, sar_dir = d["sar"], d["sar_dir"] support, resistance = d["support"], d["resistance"] trend = d["trend"] # 保本所需总费率(开+平) be_rate = fee_rate * 2 if not positions and not orders: lines.append("当前无持仓、无挂单, 账户空闲, 可按信号择机建仓.") return lines # 持仓建议 for p in positions: side = p.get("posSide", "") sz = float(p.get("pos", 0) or 0) avg = float(p.get("avgPx", 0) or 0) mark = float(p.get("markPx", 0) or 0) liq = float(p.get("liqPx", 0) or 0) upl = float(p.get("upl", 0) or 0) lev = p.get("lever", "—") # OKX net 模式(单合约净持仓)下 posSide=net 即代表多头仓位 eff_side = "long" if side in ("long", "net") else "short" side_cn = {"long": "多", "short": "空"}.get(eff_side, eff_side) # 浮盈/亏百分比(相对开仓价) pnl_pct = (mark - avg) / avg * 100 if avg else 0 if eff_side == "short": pnl_pct = -pnl_pct # 距强平价的安全垫(多头: 现价-强平; 空头: 强平-现价) if liq: if eff_side == "long": liq_buf = (last - liq) / last * 100 else: liq_buf = (liq - last) / last * 100 else: liq_buf = None # 保本价与保本最低收益率(覆盖开仓+平仓手续费, 不含资金占用时间成本) # 公式: 盈亏按名义额放大杠杆, 手续费同样按名义额计 -> 保本涨跌幅≈开平费率之和, 与杠杆无关 if eff_side == "long": be_px = avg * (1 + be_rate) be_ret = (be_px - last) / last * 100 # 现价还需再涨多少才回本 else: be_px = avg * (1 - be_rate) be_ret = (last - be_px) / last * 100 # 现价还需再跌多少才回本 adv = [] # 1) 技术面与持仓方向冲突 if eff_side == "long" and sar_dir == "空": adv.append("SAR 转空/价格已跌破 SAR, 多单与趋势背离, 建议减仓或设严止损") elif eff_side == "short" and sar_dir == "多": adv.append("SAR 转多/价格已上破 SAR, 空单与趋势背离, 建议减仓或设严止损") # 2) 逼近阻力/支撑(多单看阻力, 空单看支撑) if eff_side == "long" and (resistance - last) / last * 100 < 0.4: adv.append(f"价格贴近阻力 {resistance:.1f}, 多单可逢高部分止盈") if eff_side == "short" and (last - support) / last * 100 < 0.4: adv.append(f"价格贴近支撑 {support:.1f}, 空单可逢低部分止盈") # 3) 浮盈保护 if pnl_pct > 1.5: adv.append(f"浮盈 {pnl_pct:+.2f}%, 可上移止损至成本/保本, 锁定利润") elif pnl_pct < -1.5: adv.append(f"浮亏 {pnl_pct:+.2f}%, 接近止损阈值, 严格按原计划止损勿加仓摊平") # 4) 强平风险 if liq_buf is not None and liq_buf < 3: adv.append(f"⚠距强平价仅 {liq_buf:.1f}%, 杠杆 {lev}, 风险极高, 建议降杠杆或减仓") # 5) 保本最低收益率 if be_ret > 0: adv.append(f"保本价 {be_px:.1f}, 现价需再{'涨' if eff_side=='long' else '跌'} " f"{be_ret:.2f}% 方可覆盖手续费成本(费率 {be_rate*100:.2f}%)") else: adv.append(f"保本价 {be_px:.1f} 已低于/高于现价, 当前价位平仓已覆盖手续费成本") if not adv: adv.append("持仓方向与指标暂无明显冲突, 可持有并跟踪止损") # 盈亏率着色: 盈利红 / 亏损绿 if upl >= 0: pnl_disp = red(f"{upl:+,.2f}({pnl_pct:+.2f}%)") else: pnl_disp = green(f"{upl:+,.2f}({pnl_pct:+.2f}%)") lines.append(f"持仓[{side_cn}] 数量 {sz:.0f} 开仓 {avg:.1f} 标记 {mark:.1f} " f"浮盈亏 {pnl_disp} 保本价 {be_px:.1f} " f"强平 {liq:.1f} 杠杆 {lev}x") for a in adv: lines.append(f" ↳ {a}") # 挂单建议 for o in orders: o_side = o.get("side", "") # buy/sell o_pos = o.get("posSide", "") # long/short/net (开平方向) px = float(o.get("px", 0) or 0) sz = float(o.get("sz", 0) or 0) ord_type = o.get("ordType", "") side_cn = {"buy": "买", "sell": "卖"}.get(o_side, o_side) # 推断挂单意图: buy+net=开多挂单, sell+net=平多挂单, long/short 为双向持仓模式的开平仓 if o_pos == "net": intent = "开多" if o_side == "buy" else "平多" elif o_pos == "long": intent = "开多" if o_side == "buy" else "平多" elif o_pos == "short": intent = "开空" if o_side == "sell" else "平空" else: intent = o_pos # 挂单相对现价的位置 gap_pct = (px - last) / last * 100 adv = [] if intent in ("开多",): # 开多挂单: 通常在现价下方买入, 高于现价属突破追多 if gap_pct < -1.0: adv.append(f"挂开多单远低于现价 {gap_pct:+.2f}%, 若趋势转空可能接到下跌飞刀, 建议评估撤改") elif gap_pct > 0: adv.append(f"挂开多单高于现价 {gap_pct:+.2f}%, 属突破追多单, 确认趋势向上后再保留") else: adv.append("挂开多单贴近现价, 成交概率高, 注意仓位与止损") elif intent in ("开空",): if gap_pct > 1.0: adv.append(f"挂开空单远高于现价 {gap_pct:+.2f}%, 若趋势转多可能接到上涨飞刀, 建议评估撤改") elif gap_pct < 0: adv.append(f"挂开空单低于现价 {gap_pct:+.2f}%, 属破位追空单, 确认趋势向下后再保留") else: adv.append("挂开空单贴近现价, 成交概率高, 注意仓位与止损") elif intent in ("平多",): # 平多挂单: 通常在现价上方卖出平仓锁定利润 if gap_pct > 0.3: adv.append(f"挂平多单高于现价 {gap_pct:+.2f}%, 为高位止盈/平仓单, 趋势不破可保留") elif gap_pct < 0: adv.append(f"挂平多单低于现价 {gap_pct:+.2f}%, 属止损平仓单, 触发即离场") else: adv.append("挂平多单贴近现价, 成交概率高") elif intent in ("平空",): if gap_pct < -0.3: adv.append(f"挂平空单低于现价 {gap_pct:+.2f}%, 为低位止盈/平仓单, 趋势不破可保留") elif gap_pct > 0: adv.append(f"挂平空单高于现价 {gap_pct:+.2f}%, 属止损平仓单, 触发即离场") else: adv.append("挂平空单贴近现价, 成交概率高") # 与当前趋势一致性(仅对开仓类挂单提示) if intent == "开多" and trend == "下跌": adv.append(f"当前趋势为{trend}, 与开多挂单相悖, 谨慎") if intent == "开空" and trend == "上涨": adv.append(f"当前趋势为{trend}, 与开空挂单相悖, 谨慎") lines.append(f"挂单[{side_cn}/{intent}] 价格 {px:.1f} 数量 {sz:.0f} 类型 {ord_type} " f"(距现价 {gap_pct:+.2f}%)") for a in adv: lines.append(f" ↳ {a}") return lines # ---------- 主分析 ---------- def analyze(keys, dns_ip, fee_rate=0.0005, inst="XAU-USDT-SWAP"): c = okx_get(f"/api/v5/market/candles?instId={inst}&bar=1m&limit=120", keys, dns_ip) t = okx_get(f"/api/v5/market/ticker?instId={inst}", keys, dns_ip) if c.get("code") != "0" or t.get("code") != "0": raise RuntimeError(f"行情获取失败 c={c.get('msg')} t={t.get('msg')}") rows = c["data"][::-1] # 转升序: 最旧->最新 closes = [float(x[4]) for x in rows] highs = [float(x[2]) for x in rows] lows = [float(x[3]) for x in rows] vols = [float(x[7]) for x in rows] # USDT 计价成交量 last = closes[-1] prev = closes[-2] # 均线 sma7 = sma(closes, 7); sma25 = sma(closes, 25); sma60 = sma(closes, 60) ema12 = ema(closes, 12); ema26 = ema(closes, 26) r = rsi(closes, 14) dif, dea, hist = macd(closes) sar_v, sar_dir = sar(highs, lows) # 趋势判定 # 斜率: 近15根线性回归斜率 n = 15 ys = closes[-n:] xs = list(range(n)) mx = sum(xs) / n; my = sum(ys) / n num = sum((xs[i] - mx) * (ys[i] - my) for i in range(n)) den = sum((xs[i] - mx) ** 2 for i in range(n)) slope = num / den if den else 0 slope_pct = slope / last * 100 # 每根(分钟)百分比斜率 ema_gap = (ema12 - ema26) / last * 100 above_ma = (last > sma25) and (last > sma60) # SAR 方向加权 sar_up = (sar_dir == "多") if slope_pct > 0.02 and ema_gap > 0 and above_ma and sar_up: trend = "上涨" elif slope_pct < -0.02 and ema_gap < 0 and not above_ma and not sar_up: trend = "下跌" else: trend = "震荡" # 支撑/阻力: 近期高低 + 均线 + 枢轴 recent_h = max(highs[-60:]); recent_l = min(lows[-60:]) pivot = (recent_h + recent_l + last) / 3 # 成交量异动 avg_vol = sum(vols[-30:-1]) / 29 cur_vol = vols[-1] vol_spike = cur_vol > avg_vol * 2.2 # 关键位 support = min(recent_l, sma60 if sma60 else recent_l) resistance = max(recent_h, sma25 if sma25 else recent_h) # ---------- 操作建议逻辑 ---------- reasons = [] score = 0 # -3..+3 # 趋势权重 if trend == "上涨": score += 1; reasons.append("趋势向上(EMA12>EMA26,价格站上均线)") elif trend == "下跌": score -= 1; reasons.append("趋势向下") else: reasons.append("区间震荡, 等待突破") # RSI if r is not None: if r > 70: score -= 1; reasons.append(f"RSI {r:.0f} 超买, 追多风险大") elif r < 30: score += 1; reasons.append(f"RSI {r:.0f} 超卖, 反弹概率高") elif r > 55: reasons.append(f"RSI {r:.0f} 偏强") elif r < 45: reasons.append(f"RSI {r:.0f} 偏弱") # MACD if dif is not None: if hist > 0 and dif > dea: score += 0.5; reasons.append("MACD 金叉/红柱放大") elif hist < 0 and dif < dea: score -= 0.5; reasons.append("MACD 死叉/绿柱放大") if abs(hist) < 0.3: reasons.append("MACD 动能温和") # SAR if sar_v is not None: if sar_dir == "多": score += 0.5; reasons.append(f"SAR 多方(点 {sar_v:.1f}, 价格在其上)") else: score -= 0.5; reasons.append(f"SAR 空方(点 {sar_v:.1f}, 价格在其下)") # 位置相对支撑阻力 dist_to_s = (last - support) / last * 100 dist_to_r = (resistance - last) / last * 100 if dist_to_s < 0.3: score += 0.5; reasons.append(f"贴近支撑 {support:.1f}, 接多性价比高") elif dist_to_r < 0.3: score -= 0.5; reasons.append(f"贴近阻力 {resistance:.1f}, 做多空间受限") # 成交量异动 if vol_spike: if trend == "上涨": score += 0.5; reasons.append("放量上涨, 动量确认") elif trend == "下跌": score -= 0.5; reasons.append("放量下跌, 动量确认") else: reasons.append("成交量突增但方向未明, 警惕变盘") # 决策 if score >= 2: action = "做多"; conf = min(95, 55 + score * 12) elif score <= -2: action = "做空"; conf = min(95, 55 + abs(score) * 12) else: action = "观望"; conf = 50 + abs(score) * 5 # 入场/止损/止盈(基于建议方向, 仅供参考) if action == "做多": entry = f"{last:.1f} 附近" sl = f"{max(support, last*0.997):.1f}" tp = f"{min(resistance, last*1.006):.1f}" elif action == "做空": entry = f"{last:.1f} 附近" sl = f"{min(resistance, last*1.003):.1f}" tp = f"{max(support, last*0.994):.1f}" else: entry = sl = tp = "—" # 持仓/挂单操作建议(只读查询, 失败不影响主行情) try: positions, orders = get_positions_and_orders(keys, dns_ip, inst) pos_advice = position_advice(positions, orders, dict( last=last, sar=sar_v, sar_dir=sar_dir, support=support, resistance=resistance, trend=trend), fee_rate=fee_rate) except Exception as e: pos_advice = [f"持仓查询失败(不影响行情): {e}"] # 实时新闻(失败不影响主行情), 按标的自动关联 try: news = get_news(dns_ip, inst) except Exception: news = [] # 新闻面倾向统计(融入操作建议) nbull = sum(1 for n in news if n['bias'] == 'bull') nbear = sum(1 for n in news if n['bias'] == 'bear') if nbull > nbear: news_bias = "bull" reasons.append(f"新闻面偏多({nbull}利好/{nbear}利空{news[0]['brief'] if news else ''}), 倾向做多") if action == "做空": conf = max(20, conf - 8) elif nbear > nbull: news_bias = "bear" reasons.append(f"新闻面偏空({nbull}利好/{nbear}利空{news[0]['brief'] if news else ''}), 倾向做空") if action == "做多": conf = max(20, conf - 8) else: news_bias = "neutral" if news: reasons.append(f"新闻面中性({nbull}利好/{nbear}利空), 以技术面为主") return dict(last=last, prev=prev, sma7=sma7, sma25=sma25, sma60=sma60, ema12=ema12, ema26=ema26, rsi=r, dif=dif, dea=dea, hist=hist, sar=sar_v, sar_dir=sar_dir, trend=trend, slope_pct=slope_pct, recent_h=recent_h, recent_l=recent_l, pivot=pivot, avg_vol=avg_vol, cur_vol=cur_vol, vol_spike=vol_spike, support=support, resistance=resistance, action=action, conf=conf, entry=entry, sl=sl, tp=tp, reasons=reasons, score=score, pos_advice=pos_advice, news=news, news_bias=news_bias) def fmt(x, d=2): try: return f"{float(x):,.{d}f}" except Exception: return "—" def report(d): now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") bar = "=" * 64 inst = d.get('inst', 'XAU-USDT-SWAP') asset = news_profile(inst)['asset'] lines = [bar, f"{inst} 盯盘 {now}", bar] lines.append(f"现价 {fmt(d['last'])} 前一根 {fmt(d['prev'])} 涨跌 {(d['last']-d['prev'])/d['prev']*100:+.2f}%") lines.append(f"趋势: {d['trend']} 斜率 {d['slope_pct']:+.3f}%/min") lines.append(f"均线 SMA7 {fmt(d['sma7'])} SMA25 {fmt(d['sma25'])} SMA60 {fmt(d['sma60'])}") lines.append(f"EMA EMA12 {fmt(d['ema12'])} EMA26 {fmt(d['ema26'])} 差 {fmt((d['ema12']-d['ema26'])/d['last']*100)}%") lines.append(f"RSI(14) {fmt(d['rsi'],1)} MACD DIF {fmt(d['dif'])} DEA {fmt(d['dea'])} HIST {fmt(d['hist'])}") sar_rel = "上方(多头守护)" if (d['sar_dir'] == "多") else "下方(空方压制)" lines.append(f"SAR {fmt(d['sar'])} 方向 {d['sar_dir']} 现价{d['sar'] and ('在SAR'+sar_rel)}") lines.append(f"支撑 {fmt(d['support'])} 阻力 {fmt(d['resistance'])} 枢轴 {fmt(d['pivot'])}") lines.append(f"成交量 当前 {fmt(d['cur_vol'])} USDT / 均值 {fmt(d['avg_vol'])} {'⚡异动' if d['vol_spike'] else ''}") lines.append("-" * 64) lines.append(f"【操作建议】 {d['action']} 置信度 {d['conf']:.0f}%") if d['action'] != "观望": lines.append(f" 入场 {d['entry']} 止损 {d['sl']} 止盈 {d['tp']} (3~15分钟级别)") lines.append(" 依据:") for rsn in d['reasons']: lines.append(f" · {rsn}") lines.append("-" * 64) lines.append("【当前持仓操作建议】(基于实时行情, 仅供参考, 脚本不下单)") for pa in d.get('pos_advice', []): lines.append(f" {pa}") lines.append("-" * 64) lines.append(f"【相关新闻 · {asset}】(金十 7x24, 实时抓取, 按标的自动关联)") news = d.get('news', []) if not news: lines.append(f" 暂未抓取到与{asset}相关快讯") else: bias_cn = {"bull": red(f"利好{asset}"), "bear": green(f"利空{asset}"), "neutral": "中性影响"} for n in news: lines.append(f" · {n['brief']} [{bias_cn.get(n['bias'], '中性影响')}]") # 新闻对持仓操作的综合倾向提示 bull = sum(1 for n in news if n['bias'] == 'bull') bear = sum(1 for n in news if n['bias'] == 'bear') bias_word = {"bull": red("偏多"), "bear": green("偏空"), "neutral": "中性"}[d.get('news_bias', 'neutral')] lines.append(f" ↳ 新闻面{bias_word}({bull}利好/{bear}利空), 已融入上方操作建议") lines.append(bar) return "\n".join(lines) def log_write(text): with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(strip_ansi(text) + "\n") def popup(title, msg): try: import ctypes ctypes.windll.user32.MessageBoxW(0, msg, title, 0x40 | 0x1000) except Exception: pass def main(): ap = argparse.ArgumentParser() ap.add_argument("--dns-ip", default="198.18.0.0") ap.add_argument("--once", action="store_true") ap.add_argument("--alert-popup", action="store_true", help="行情突变弹窗提醒") ap.add_argument("--fee-rate", type=float, default=0.0005, help="单边 taker 手续费率(默认 0.0005=0.05%, 保本按开+平两次计)") ap.add_argument("--inst", default="XAU-USDT-SWAP", help="盯盘标的代码, 如 XAU-USDT-SWAP / BTC-USDT-SWAP / ETH-USDT-SWAP") args = ap.parse_args() enable_color() # 启用终端 ANSI 颜色(盈亏率着色) keys = load_keys() inst = args.inst.upper() last_alert = {} # 突变去重 while True: try: d = analyze(keys, args.dns_ip, fee_rate=args.fee_rate, inst=inst) d['inst'] = inst out = report(d) print(out) log_write(out) # 行情突变提醒 price_chg = abs(d['last'] - d['prev']) / d['prev'] * 100 alert_msgs = [] if d['vol_spike']: alert_msgs.append(f"⚡成交量突增: 当前 {fmt(d['cur_vol'])} vs 均值 {fmt(d['avg_vol'])}") if price_chg > 0.3: alert_msgs.append(f"⚠价格急动: {d['last']-d['prev']:+.1f} ({price_chg:+.2f}%)/分钟") if d['rsi'] is not None and (d['rsi'] > 75 or d['rsi'] < 25): alert_msgs.append(f"⚠RSI 极端 {d['rsi']:.0f} (超{'买' if d['rsi']>75 else '卖'})") if alert_msgs and (args.alert_popup or True): tag = "|".join(alert_msgs) if last_alert.get(tag, 0) < time.time() - 300: # 5分钟内同内容不重复 log_write("【提醒】 " + " ; ".join(alert_msgs)) if args.alert_popup: popup("XAU 行情突变", "\n".join(alert_msgs) + f"\n现价 {fmt(d['last'])}") last_alert[tag] = time.time() except Exception as e: err = f"[{datetime.datetime.now()}] 运行错误: {e}" print(err); log_write(err) if args.once: break # ---------- 刷新倒计时(分:秒), 归零自动触发下一轮刷新 ---------- # 用目标绝对时间点计算剩余, 避免 sleep 累积漂移导致计时不同步 next_at = time.time() + INTERVAL while True: remain = int(round(next_at - time.time())) if remain <= 0: break mm = remain // 60 ss = remain % 60 # \r 回到行首覆盖, 不换行; flush 确保即时显示 print(f"\r距下次刷新: {mm:02d}:{ss:02d} ", end="", flush=True) # 整秒粒度休眠, 剩余 <1s 时立即退出循环进入刷新 time.sleep(min(1, remain)) # 收尾换行, 下一轮刷新会先打印分隔报告 print("\r距下次刷新: 00:00 刷新中...\n", flush=True) if __name__ == "__main__": main()