Hermes-Skills/intraday-regime-detector/scripts/regime_scan.py

#!/usr/bin/env python3
"""
regime_scan.py - 扫描港美股日内候选的市场状态 + 推荐策略
不交易, 只判别 + 推 QQ
"""
import sys
import json
import subprocess
import re
from pathlib import Path

sys.path.insert(0, '/home/openclaw/.hermes/skills/trading/intraday-regime-detector/scripts')
from intraday_regime import IntradayStrategySelector, MarketRegime, StrategyType

CANDIDATE_HK = Path('/home/openclaw/.hermes/skills/trading/quant-factor-mining/artifacts/hk_intraday_latest.json')
CANDIDATE_US = Path('/home/openclaw/.hermes/skills/trading/quant-factor-mining/artifacts/us_intraday_latest.json')


def fetch_klines_hk(symbol: str, period: str = '5m', count: int = 30) -> list:
    """港股表格 parser"""
    result = subprocess.run(
        ['proxychains4', '-f', '/home/openclaw/.proxychains/proxychains.conf',
         '/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real',
         '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({
                'open': parse_num(o),
                'high': parse_num(h),
                'low': parse_num(l),
                'close': parse_num(c),
                'volume': parse_num(v),
            })
    return klines


def fetch_klines_us(symbol: str, period: str = '5m', count: int = 30) -> list:
    """美股 JSON"""
    result = subprocess.run(
        ['proxychains4', '-f', '/home/openclaw/.proxychains/proxychains.conf',
         '/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real',
         'candlesticks', symbol, period, '--count', str(count), '--json'],
        capture_output=True, text=True, timeout=30,
    )
    start = result.stdout.find('[')
    if start == -1:
        return []
    try:
        data = json.loads(result.stdout[start:])
        return [{
            'open': float(k['open']),
            'high': float(k['high']),
            'low': float(k['low']),
            'close': float(k['close']),
            'volume': float(k.get('volume', 0)),
        } for k in data if 'close' in k]
    except Exception:
        return []


def fetch_quote(symbol: str) -> dict:
    result = subprocess.run(
        ['proxychains4', '-f', '/home/openclaw/.proxychains/proxychains.conf',
         '/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real',
         'quote', symbol, '--json'],
        capture_output=True, text=True, timeout=30,
    )
    text = result.stdout
    start = text.find('[')
    if start == -1:
        return {}
    try:
        return json.loads(text[start:])[0]
    except Exception:
        return {}


def analyze_market(symbol: str, market: str, klines: list, quote: dict) -> str:
    """返回单支票分析报告"""
    if not klines or not quote:
        return f"❌ {symbol} 数据缺失"

    try:
        import pandas as pd
        df = pd.DataFrame(klines)
    except ImportError:
        return f"❌ pandas 未装"

    prev_close = quote.get('prev_close', 0)
    current_price = quote['last_done']
    open_p = quote['open']
    gap_pct = ((open_p - prev_close) / prev_close * 100) if prev_close else 0

    selector = IntradayStrategySelector()
    diag = selector.diagnose(df, open_gap_pct=gap_pct)

    # 策略 emoji
    strategy_emoji = {
        StrategyType.TREND_FOLLOWING: '📈',
        StrategyType.GRID_TRADING: '🔲',
        StrategyType.BOLLINGER_REVERSION: '📊',
        StrategyType.NO_TRADE: '⛔',
    }
    regime_short = {
        MarketRegime.STRONG_TREND_UP: '强趋↑',
        MarketRegime.STRONG_TREND_DOWN: '强趋↓',
        MarketRegime.HIGH_VOL_SHAKE: '高波震荡',
        MarketRegime.LOW_VOL_STABLE: '低波震荡',
        MarketRegime.CHAOTIC: '混乱',
        MarketRegime.UNKNOWN: '未知',
    }

    params_str = '\n'.join(f"  {k}: {v}" for k, v in diag.strategy_params.items())

    return (
        f"\n{strategy_emoji.get(diag.recommended_strategy, '•')} **{symbol}** ({market}) "
        f"现价 ${current_price:.2f} ({gap_pct:+.2f}%) "
        f"置信度 {diag.confidence:.0%}\n"
        f"  状态: {regime_short.get(diag.regime, diag.regime.value)} | "
        f"R²={diag.r_squared} | 波动率={diag.volatility:.2%} | ADF p={diag.adf_pvalue}\n"
        f"  推荐: {diag.recommended_strategy.value}\n"
        f"{params_str}"
    )


def scan_market(market: str, candidate_file: Path, fetch_klines_func) -> list:
    """扫描一个市场"""
    if not candidate_file.exists():
        return [f"⚠️ 候选池不存在: {candidate_file.name}"]

    with open(candidate_file) as f:
        data = json.load(f)

    results = data.get('results', [])[:5]  # top 5
    date = data.get('date', '?')[:10]

    if not results:
        return [f"⚠️ {market} 候选池为空"]

    reports = [f"📊 {market} 日内市场状态扫描 ({date})"]

    for entry in results:
        symbol = entry['ticker']
        score = entry['score']
        try:
            quote = fetch_quote(symbol)
            klines = fetch_klines_func(symbol, '5m', 30)
            report = analyze_market(symbol, market, klines, quote)
            reports.append(report)
        except Exception as e:
            reports.append(f"❌ {symbol} 异常: {e}")

    return reports


def main():
    # 港股 + 美股
    hk_reports = scan_market('HK', CANDIDATE_HK, fetch_klines_hk)
    us_reports = scan_market('US', CANDIDATE_US, fetch_klines_us)

    print(f"📊 日内市场状态扫描 ({hk_reports[0].split('(')[-1].rstrip(')')})\n")
    print('=' * 60)

    print('\n--- 港股 ---')
    for r in hk_reports[1:]:
        print(r)
    print()

    print('\n--- 美股 ---')
    for r in us_reports[1:]:
        print(r)


if __name__ == '__main__':
    main()