294 lines
11 KiB
Python
294 lines
11 KiB
Python
---
|
|
name: hk-scan
|
|
description: "港股日内做T点位扫描 — 拉候选池 top 5 + 实时 quote + 5min K, 算 SL/TP1/TP2 推 QQ (不交易, 仅参考)"
|
|
---
|
|
|
|
"""
|
|
港股日内做T点位扫描 (cron 模板)
|
|
- 拉候选池 top 5 (artifact hk_intraday_latest.json)
|
|
- 拉实时 quote + 5min K 线
|
|
- 算 SL/TP1/TP2 用 calc_exit_levels()
|
|
- 三级输出: ✅ R:R≥1.5 / ⚠️ R:R 1.0 / ❌ 否决
|
|
- 推 QQ (origin delivery)
|
|
|
|
用法:
|
|
python3 calc_hk_levels.py # 跑 (cron 默认)
|
|
python3 calc_hk_levels.py --top 3 # 只看 top 3
|
|
python3 calc_hk_levels.py --period 60m # 用 60min K
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 添加 strategy-management scripts 到 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/hk_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_hk(symbol: str, period: str = '5m', count: int = 30) -> list:
|
|
"""
|
|
港股 K 线 (表格 parser, 6 列, 不用 --json)
|
|
- 分隔符 │ (U+2502), 不是 |
|
|
- 表头中文: 时间/开盘/最高/最低/收盘/成交量
|
|
- 时间格式: '2026-07-10 10:30' (空格分隔)
|
|
- 数字带千分位逗号: '1,190,430'
|
|
"""
|
|
result = subprocess.run(
|
|
PROXYCHAINS + LONGBRIDGE + ['candlesticks', symbol, period, '--count', str(count)],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
klines = []
|
|
pattern = re.compile(
|
|
r'│\s*(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2})\s*│'
|
|
r'\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│'
|
|
)
|
|
for line in result.stdout.split('\n'):
|
|
m = pattern.search(line)
|
|
if m:
|
|
ts, o, h, l, c, v = m.groups()
|
|
def parse_num(s):
|
|
return float(s.replace(',', ''))
|
|
klines.append({
|
|
'timestamp': ts.replace(' ', 'T'),
|
|
'open': parse_num(o),
|
|
'high': parse_num(h),
|
|
'low': parse_num(l),
|
|
'close': parse_num(c),
|
|
'volume': parse_num(v),
|
|
})
|
|
return 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['volume'] 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':
|
|
vol_sl_multi, vol_tp1_multi, vol_tp2_multi = 2.0, 4.0, 8.0
|
|
eff_min_rr = min(min_rr, 1.0)
|
|
use_vwap = False
|
|
elif strategy == 'vwap_revert':
|
|
vol_sl_multi, vol_tp1_multi, vol_tp2_multi = 1.0, 2.0, 3.0
|
|
eff_min_rr = max(min_rr, 1.5)
|
|
use_vwap = True
|
|
elif strategy == 'early_bird':
|
|
vol_sl_multi, vol_tp1_multi, vol_tp2_multi = 1.5, 2.0, 3.0
|
|
eff_min_rr = min_rr
|
|
use_vwap = False
|
|
else: # rsi2_revert
|
|
vol_sl_multi, vol_tp1_multi, vol_tp2_multi = 1.5, 2.0, 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):
|
|
"""格式化为 QQ 推送文本 (单条)"""
|
|
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: # atr_adj
|
|
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, help='候选池 top N (default 5)')
|
|
ap.add_argument('--period', default='5m', help='K 线周期 (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_hk(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'
|
|
|
|
# 三级尝试: 严格 / 宽松 / ATR 调整
|
|
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:
|
|
# 场景 B: 宽松
|
|
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:
|
|
# 场景 C: ATR 倍数调整 (SL=1.5 ATR, TP1=3.0 ATR)
|
|
closes = [k['close'] for k in klines]
|
|
highs = [k['high'] for k in klines]
|
|
lows = [k['low'] for k in klines]
|
|
volumes = [k['volume'] 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()
|