--- name: us-scan description: "美股日内做T点位扫描 — 拉候选池 top 5 + 实时 quote + 5min K, 算 SL/TP1/TP2 推 QQ (不交易, 仅参考)" --- """ 美股日内做T点位扫描 (cron 模板) - 拉候选池 top 5 (artifact us_intraday_latest.json) - 拉实时 quote + 5min K 线 (JSON 输出) - 算 SL/TP1/TP2 用 calc_exit_levels() - 三级输出: ✅ R:R≥1.5 / ⚠️ R:R 1.0 / ❌ 否决 - 推 QQ (origin delivery) 用法: python3 calc_us_levels.py # 跑 (cron 默认) python3 calc_us_levels.py --top 3 # 只看 top 3 python3 calc_us_levels.py --period 60m # 用 60min K """ import argparse import json import os import subprocess import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from exit_levels import calc_exit_levels from indicators import atr as calc_atr, vwap as calc_vwap # === 常量 === HERMES_HOME = '/home/openclaw' CANDIDATE_FILE = f'{HERMES_HOME}/.hermes/skills/trading/quant-factor-mining/artifacts/us_intraday_latest.json' PROXYCHAINS = ['proxychains4', '-f', f'{HERMES_HOME}/.proxychains/proxychains.conf'] LONGBRIDGE = ['/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real'] # === 长桥数据拉取 (美股版) === def fetch_quote(symbol: str) -> dict: """美股 quote, JSON""" result = subprocess.run( PROXYCHAINS + LONGBRIDGE + ['quote', symbol, '--json'], capture_output=True, text=True, timeout=30, ) start = result.stdout.find('[') if start == -1: return {} try: return json.loads(result.stdout[start:])[0] except Exception: return {} def fetch_klines_us(symbol: str, period: str = '5m', count: int = 30) -> list: """ 美股 K 线 (JSON, 7 列含 turnover) - 支持 --json - stdout 拼接表格提示 + JSON → find('[') 切 - 时间格式: '2026-07-16T03:55:00' (T 分隔) - 数字无千分位 """ result = subprocess.run( PROXYCHAINS + LONGBRIDGE + ['candlesticks', symbol, period, '--count', str(count), '--json'], capture_output=True, text=True, timeout=30, ) # 长桥把表格提示 (USOption/HK) + JSON 拼一起 start = result.stdout.find('[') if start == -1: return [] try: data = json.loads(result.stdout[start:]) return [k for k in data if 'timestamp' in k and 'close' in k] except Exception: return [] # === 业务逻辑 (与 calc_hk_levels.py 完全一致, 只是 klines 函数不同) === def calc_levels(symbol: str, klines: list, quote: dict, side: str, min_rr: float, strategy: str = 'rsi2_revert'): """ strategy: 'rsi2_revert' | 'vwap_revert' | 'early_bird' | 'turtle_breakout' 每个策略用不同的 vol_multi 参数组 """ if not klines or not quote: return None, '数据缺失' closes = [k['close'] for k in klines] highs = [k['high'] for k in klines] lows = [k['low'] for k in klines] volumes = [k.get('volume', 0) for k in klines] atr_vals = calc_atr(highs, lows, closes, 14) current_atr = atr_vals[-1] if not current_atr: return None, 'ATR 失败' vwaps = calc_vwap(closes, volumes) current_vwap = vwaps[-1] current_price = quote['last_done'] # ── 策略参数映射 ── if strategy == 'turtle_breakout': # 海龟: SL=2ATR, TP1=4ATR(1R), TP2=8ATR(2R), 更宽松 min_rr=1.0 vol_sl_multi = 2.0 vol_tp1_multi = 4.0 vol_tp2_multi = 8.0 eff_min_rr = min(min_rr, 1.0) use_vwap = False # 海龟不看 VWAP elif strategy == 'vwap_revert': # VWAP 回归: SL=1σ, TP1=VWAP±0.05%, TP2=2ATR vol_sl_multi = 1.0 vol_tp1_multi = 2.0 vol_tp2_multi = 3.0 eff_min_rr = max(min_rr, 1.5) use_vwap = True elif strategy == 'early_bird': # 开盘缺口: SL=1.5ATR, TP1=2ATR, TP2=3ATR vol_sl_multi = 1.5 vol_tp1_multi = 2.0 vol_tp2_multi = 3.0 eff_min_rr = min_rr use_vwap = False else: # rsi2_revert (默认): SL=1.5, TP1=2.0, TP2=3.0 vol_sl_multi = 1.5 vol_tp1_multi = 2.0 vol_tp2_multi = 3.0 eff_min_rr = min_rr use_vwap = True eff_vwap = current_vwap if use_vwap else None return calc_exit_levels( entry=current_price, atr=current_atr, current_price=current_price, day_high=quote['high'], day_low=quote['low'], prev_high=max(highs), prev_low=min(lows), vwap=eff_vwap, side=side, min_rr=eff_min_rr, vol_sl_multi=vol_sl_multi, vol_tp1_multi=vol_tp1_multi, vol_tp2_multi=vol_tp2_multi, ), strategy def _read_policy(path): p = Path(path) return p.read_text(encoding="utf-8").strip() if p.exists() else None def format_qq_output(levels, change, current_price, side, symbol, score, adr, mode): if mode == 'strict': return ( f"\n📈 **{symbol}** (score {score}, ADR {adr}%)\n" f"现价 ${current_price:.2f} ({change:+.2f}%) | {side.upper()}\n" f"SL ${levels.sl:.2f} ({levels.sl_method})\n" f"TP1 ${levels.tp1:.2f} ({levels.tp_method})\n" f"TP2 ${levels.tp2:.2f}\n" f"R:R 1:{levels.rr_ratio:.2f} ✅" ) elif mode == 'relaxed': return ( f"\n📈 **{symbol}** (score {score}, ADR {adr}%)\n" f"现价 ${current_price:.2f} ({change:+.2f}%) | {side.upper()} [R:R 1.0 宽松]\n" f"SL ${levels.sl:.2f}\n" f"TP1 ${levels.tp1:.2f}\n" f"TP2 ${levels.tp2:.2f}\n" f"R:R 1:{levels.rr_ratio:.2f} ⚠️" ) else: return ( f"\n📈 **{symbol}** (score {score}, ADR {adr}%)\n" f"现价 ${current_price:.2f} ({change:+.2f}%) | {side.upper()} [ATR 调整]\n" f"SL ${levels.sl:.2f}\n" f"TP1 ${levels.tp1:.2f}\n" f"TP2 ${levels.tp2:.2f}\n" f"R:R 1:{levels.rr_ratio:.2f} ⚠️" ) def main(): ap = argparse.ArgumentParser() ap.add_argument('--top', type=int, default=5) ap.add_argument('--period', default='5m') ap.add_argument('--strategy', default='rsi2_revert', choices=['rsi2_revert', 'vwap_revert', 'early_bird', 'turtle_breakout'], help='策略 (default rsi2_revert)') ap.add_argument('--policy-cn', default='/tmp/policy_cn.txt', help='国内政策文件 (空/不存在则跳过)') ap.add_argument('--policy-intl', default='/tmp/policy_intl.txt', help='国际政策文件 (空/不存在则跳过)') args = ap.parse_args() if not os.path.exists(CANDIDATE_FILE): print(f"[skip] 候选池不存在: {CANDIDATE_FILE}") return with open(CANDIDATE_FILE) as f: candidate_data = json.load(f) top = candidate_data.get('results', [])[:args.top] date = candidate_data.get('date', '?')[:10] print(f"📊 美股日内做T点位扫描 (候选池 {date}, top {args.top})") print(f"📋 共扫描 {len(top)} 支\n") output_lines = [] for entry in top: symbol = entry['ticker'] score = entry['score'] avg_adr = entry['avg_adr'] print(f"--- {symbol} (score {score}, ADR {avg_adr}%) ---") quote = fetch_quote(symbol) if not quote: print(f" ❌ quote 拉取失败") continue klines = fetch_klines_us(symbol, args.period, 30) if not klines: print(f" ❌ K线 拉取失败") continue current_price = quote['last_done'] change = (current_price - quote['prev_close']) / quote['prev_close'] * 100 print(f" 现价: ${current_price:.2f} ({change:+.2f}%)") side = 'long' if change > 0 else 'short' levels, _ = calc_levels(symbol, klines, quote, side, min_rr=1.5, strategy=args.strategy) if levels: print(f" ✅ R:R 1.5 [{args.strategy}] → SL=${levels.sl:.2f} TP1=${levels.tp1:.2f} TP2=${levels.tp2:.2f} R:R=1:{levels.rr_ratio:.2f}") output_lines.append(format_qq_output(levels, change, current_price, side, symbol, score, avg_adr, 'strict')) else: levels_relaxed, _ = calc_levels(symbol, klines, quote, side, min_rr=1.0, strategy=args.strategy) if levels_relaxed: print(f" ⚠️ R:R 1.5 否决, 1.0 通过 → R:R=1:{levels_relaxed.rr_ratio:.2f}") output_lines.append(format_qq_output(levels_relaxed, change, current_price, side, symbol, score, avg_adr, 'relaxed')) else: # ATR 调整 closes = [k['close'] for k in klines] highs = [k['high'] for k in klines] lows = [k['low'] for k in klines] volumes = [k.get('volume', 0) for k in klines] atr_v = calc_atr(highs, lows, closes, 14)[-1] vwaps = calc_vwap(closes, volumes) levels_alt = calc_exit_levels( entry=current_price, atr=atr_v, current_price=current_price, day_high=quote['high'], day_low=quote['low'], prev_high=max(highs), prev_low=min(lows), vwap=vwaps[-1], side=side, min_rr=1.5, vol_sl_multi=1.5, vol_tp1_multi=3.0, ) if levels_alt: print(f" ⚠️ ATR 调整 → R:R=1:{levels_alt.rr_ratio:.2f}") output_lines.append(format_qq_output(levels_alt, change, current_price, side, symbol, score, avg_adr, 'atr_adj')) else: print(f" ❌ 全部场景否决") print() if output_lines: header_parts = [] pol_cn = _read_policy(args.policy_cn) pol_intl = _read_policy(args.policy_intl) if pol_cn: header_parts.append(pol_cn) if pol_intl: header_parts.append(pol_intl) header_parts.append(f"📊 美股日内做T点位 ({date})") if args.strategy != 'rsi2_revert': header_parts.append(f"(策略: {args.strategy})") header_parts.append("⚠️ 仅参考, 不交易") header = "\n".join(header_parts) print("\n=== QQ 推送内容 ===") print(header + "\n" + "\n---\n".join(output_lines)) else: print("\n💤 全部场景否决, 无输出") if __name__ == '__main__': main()