31 lines
1.1 KiB
Python
31 lines
1.1 KiB
Python
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""信号冷却过滤模块
|
|||
|
|
- 避免同一标的在冷却期内重复产生同方向信号
|
|||
|
|
- 过滤后附加 generated_at_utc 字段（UTC ISO）
|
|||
|
|
"""
|
|||
|
|
from datetime import datetime, timezone, timedelta
|
|||
|
|
from typing import List, Dict, Any
|
|||
|
|
|
|||
|
|
class SignalCooldownFilter:
|
|||
|
|
def __init__(self, cooldown_minutes: int = 30):
|
|||
|
|
self.cooldown = timedelta(minutes=cooldown_minutes)
|
|||
|
|
# key: (symbol, direction) -> last datetime
|
|||
|
|
self.last_time: Dict[tuple, datetime] = {}
|
|||
|
|
|
|||
|
|
def filter(self, signals: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|||
|
|
now = datetime.now(timezone.utc)
|
|||
|
|
accepted = []
|
|||
|
|
for s in signals:
|
|||
|
|
symbol = s.get('symbol')
|
|||
|
|
direction = s.get('type') or s.get('direction')
|
|||
|
|
key = (symbol, direction)
|
|||
|
|
lt = self.last_time.get(key)
|
|||
|
|
if lt is not None and now - lt < self.cooldown:
|
|||
|
|
continue
|
|||
|
|
# Accept
|
|||
|
|
self.last_time[key] = now
|
|||
|
|
s['direction'] = direction # normalize
|
|||
|
|
s['generated_at_utc'] = now.strftime('%Y-%m-%dT%H:%M:%SZ')
|
|||
|
|
accepted.append(s)
|
|||
|
|
return accepted
|