cyb50-quant/kalman-filter/test_extract.py

#!/usr/bin/env python3
import subprocess

result = subprocess.run(['python3', 'cyb50_kalman_filter_daily.py'], cwd='/root/.openclaw/workspace/kalman-filter', capture_output=True, text=True, timeout=300)

output = result.stdout
lines = output.split('\n')

signals = []
in_signal_section = False

for line in lines:
    if '所有交易信号详情' in line:
        in_signal_section = True
        continue
    if in_signal_section:
        stripped = line.strip()
        if len(stripped) > 30:
            parts = stripped.split()
            if len(parts) >= 3 and parts[0].isdigit() and len(parts[1]) == 10 and parts[1][4] == '-':
                signals.append({
                    'num': parts[0],
                    'date': parts[1],
                    'action': parts[2],
                    'price': parts[3] if len(parts) > 3 else '',
                })
        # 获取所有信号
        if '当前市场状态' in line:
            break

print(f'找到 {len(signals)} 个交易信号')

# 取最近40个信号
recent_signals = signals[-40:] if len(signals) > 40 else signals
print(f'\n最近40个信号中的前10个:')
for s in recent_signals[:10]:
    print(f"  {s['num']:>3} {s['date']} {s['action']:>4} {s['price']}")

print(f'\n最近40个信号中的后10个:')
for s in recent_signals[-10:]:
    print(f"  {s['num']:>3} {s['date']} {s['action']:>4} {s['price']}")

# 配对
trades = []
for i in range(len(recent_signals) - 1):
    if recent_signals[i]['action'] == '买入' and recent_signals[i+1]['action'] == '卖出':
        buy = recent_signals[i]
        sell = recent_signals[i+1]
        trades.append(f"{buy['num']} {buy['date']}买入→{sell['date']}卖出")

print(f'\n配对到 {len(trades)} 对交易')
print('\n最近5对:')
for t in trades[-5:]:
    print(f'  {t}')