Hermes-Skills/intraday-trading/scripts/hk_intraday_scanner.py

#!/usr/bin/env python3
"""港股日内交易盘前筛选 - 8:30自动运行"""
import os, json
from datetime import datetime

# Load LongBridge credentials
config = {}
with open(os.path.expanduser('~/.bashrc'), 'r') as f:
    for line in f:
        if line.startswith('export LONGPORT_'):
            key, value = line.strip().split('=', 1)
            config[key.replace('export ', '')] = value

os.environ['LONGPORT_APP_KEY'] = config.get('LONGPORT_APP_KEY', '')
os.environ['LONGPORT_APP_SECRET'] = config.get('LONGPORT_APP_SECRET', '')
os.environ['LONGPORT_ACCESS_TOKEN'] = config.get('LONGPORT_ACCESS_TOKEN', '')

from longport import openapi

cfg = openapi.Config.from_env()
ctx = openapi.QuoteContext(config=cfg)

# 候选标的池
tickers = [
    '700.HK', '9988.HK', '1810.HK', '3690.HK', '9888.HK',
    '9618.HK', '1024.HK', '2015.HK', '9866.HK', '9868.HK',
    '5.HK', '388.HK', '1299.HK', '2318.HK', '1398.HK',
]

quotes = ctx.quote(tickers)
indexes = ctx.calc_indexes(tickers, [
    openapi.CalcIndex.VolumeRatio, openapi.CalcIndex.TurnoverRate,
])

results = []
for ticker in tickers:
    try:
        candles = ctx.candlesticks(ticker, openapi.Period.Day, 20, openapi.AdjustType.ForwardAdjust)
        if not candles:
            continue
        highs = [float(c.high) for c in candles]
        lows = [float(c.low) for c in candles]
        closes = [float(c.close) for c in candles]
        adrs = [(h - l) / c * 100 for h, l, c in zip(highs, lows, closes)]
        avg_adr = sum(adrs[-5:]) / 5  # 近5日ADR
        q = next((q for q in quotes if q.symbol == ticker), None)
        idx = next((i for i in indexes if i.symbol == ticker), None)
        if q and idx:
            vr = float(getattr(idx, 'volume_ratio', 0) or 0)
            tr = float(getattr(idx, 'turnover_rate', 0) or 0)
            # 评分：ADR 40% + 量比 30% + 换手率 30%
            score = min(avg_adr / 4, 1) * 40 + min(vr / 2, 1) * 30 + min(tr / 2, 1) * 30
            results.append({
                'ticker': ticker, 'price': float(q.last_done),
                'volume_ratio': vr, 'turnover_rate': tr,
                'avg_adr': round(avg_adr, 2), 'score': round(score, 1),
            })
    except Exception as e:
        continue

results.sort(key=lambda x: x['score'], reverse=True)

# 保存结果
out_path = os.path.expanduser('~/.hermes/skills/trading/quant-factor-mining/artifacts/hk_intraday_latest.json')
os.makedirs(os.path.dirname(out_path), exist_ok=True)
with open(out_path, 'w') as f:
    json.dump({'date': datetime.now().isoformat(), 'results': results[:8]}, f, ensure_ascii=False, indent=2)

# 输出报告
date_str = datetime.now().strftime('%Y-%m-%d')
print(f'🔥 港股日内交易盘前筛选 {date_str}')
print('=' * 55)
print(f'{"股票":<10}{"现价":>8}{"ADR%":>7}{"量比":>6}{"换手":>6}{"评分":>6}')
print('-' * 55)

for r in results[:8]:
    emoji = '🟢' if r['score'] > 60 else ('🟡' if r['score'] > 40 else '🔴')
    print(f'{emoji}{r["ticker"]:<9}{r["price"]:>8.2f}{r["avg_adr"]:>7.2f}{r["volume_ratio"]:>6.2f}{r["turnover_rate"]:>6.2f}{r["score"]:>6.1f}')

print()
print('📋 TOP 3 策略建议:')
for r in results[:3]:
    if r['avg_adr'] > 4:
        strategy = '动量突破'
    elif r['avg_adr'] > 3:
        strategy = '趋势跟踪'
    else:
        strategy = 'VWAP回归'
    print(f'  {r["ticker"]}: {strategy} | 止损-1.5% | 量比{r["volume_ratio"]:.1f}')