312 lines
9.3 KiB
Python
312 lines
9.3 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
📰 NEWS FILTER SIMPLE - Manual Event Configuration
|
||
|
|
Blockiert Trading vor/nach High-Impact News Events
|
||
|
|
|
||
|
|
NO API NEEDED - Manual event list configuration
|
||
|
|
|
||
|
|
FEATURES:
|
||
|
|
- Simple JSON configuration
|
||
|
|
- 30min Buffer vor/nach Event
|
||
|
|
- Easy to maintain event list
|
||
|
|
- Works offline
|
||
|
|
|
||
|
|
CONFIGURATION:
|
||
|
|
Edit news_events_manual.json and add upcoming events:
|
||
|
|
{
|
||
|
|
"events": [
|
||
|
|
{
|
||
|
|
"date": "2026-01-10",
|
||
|
|
"time": "13:30",
|
||
|
|
"timezone": "UTC",
|
||
|
|
"event": "US NFP (Non-Farm Payrolls)",
|
||
|
|
"currency": "USD",
|
||
|
|
"impact": "HIGH"
|
||
|
|
}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
|
||
|
|
VERWENDUNG:
|
||
|
|
from news_filter_simple import is_news_upcoming
|
||
|
|
|
||
|
|
if is_news_upcoming(minutes_ahead=30):
|
||
|
|
print("⏸️ Trading SKIP: High-Impact News Event")
|
||
|
|
return
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from typing import List, Dict
|
||
|
|
|
||
|
|
|
||
|
|
class SimpleNewsFilter:
|
||
|
|
"""
|
||
|
|
Simple News Filter with manual event configuration
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, config_file: str = "news_events_manual.json"):
|
||
|
|
"""
|
||
|
|
Initialize Simple News Filter
|
||
|
|
|
||
|
|
Args:
|
||
|
|
config_file: Path to events config file
|
||
|
|
"""
|
||
|
|
self.config_file = config_file
|
||
|
|
self.events = []
|
||
|
|
self._load_events()
|
||
|
|
|
||
|
|
|
||
|
|
def _load_events(self):
|
||
|
|
"""Load events from config file"""
|
||
|
|
if not os.path.exists(self.config_file):
|
||
|
|
self._create_template()
|
||
|
|
return
|
||
|
|
|
||
|
|
try:
|
||
|
|
with open(self.config_file, 'r') as f:
|
||
|
|
data = json.load(f)
|
||
|
|
self.events = data.get('events', [])
|
||
|
|
|
||
|
|
print(f"✅ Loaded {len(self.events)} news events from {self.config_file}")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"⚠️ Error loading events: {e}")
|
||
|
|
self.events = []
|
||
|
|
|
||
|
|
|
||
|
|
def _create_template(self):
|
||
|
|
"""Create template configuration file"""
|
||
|
|
template = {
|
||
|
|
"events": [
|
||
|
|
{
|
||
|
|
"date": "2026-01-10",
|
||
|
|
"time": "13:30",
|
||
|
|
"timezone": "UTC",
|
||
|
|
"event": "US NFP (Non-Farm Payrolls)",
|
||
|
|
"currency": "USD",
|
||
|
|
"impact": "HIGH",
|
||
|
|
"note": "First Friday of every month"
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"date": "2026-01-15",
|
||
|
|
"time": "13:30",
|
||
|
|
"timezone": "UTC",
|
||
|
|
"event": "US CPI (Consumer Price Index)",
|
||
|
|
"currency": "USD",
|
||
|
|
"impact": "HIGH",
|
||
|
|
"note": "Mid-month, around 13th-15th"
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"date": "2026-01-29",
|
||
|
|
"time": "19:00",
|
||
|
|
"timezone": "UTC",
|
||
|
|
"event": "FOMC Interest Rate Decision",
|
||
|
|
"currency": "USD",
|
||
|
|
"impact": "HIGH",
|
||
|
|
"note": "Check Fed calendar for actual date"
|
||
|
|
}
|
||
|
|
],
|
||
|
|
"instructions": [
|
||
|
|
"Add upcoming high-impact events here",
|
||
|
|
"Common events: NFP (1st Friday), CPI (mid-month), FOMC (8 times/year)",
|
||
|
|
"Time format: HH:MM in 24-hour format",
|
||
|
|
"Date format: YYYY-MM-DD",
|
||
|
|
"Always use UTC timezone"
|
||
|
|
]
|
||
|
|
}
|
||
|
|
|
||
|
|
with open(self.config_file, 'w') as f:
|
||
|
|
json.dump(template, f, indent=2)
|
||
|
|
|
||
|
|
print(f"✅ Created template: {self.config_file}")
|
||
|
|
print(f"📝 Edit this file and add your upcoming news events")
|
||
|
|
|
||
|
|
|
||
|
|
def get_upcoming_news(self, minutes_ahead: int = 60, minutes_after: int = 30) -> List[Dict]:
|
||
|
|
"""
|
||
|
|
Get upcoming news events within time window
|
||
|
|
|
||
|
|
Args:
|
||
|
|
minutes_ahead: Look ahead this many minutes
|
||
|
|
minutes_after: Consider events from this many minutes ago
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
List of upcoming events with minutes_until
|
||
|
|
"""
|
||
|
|
now = datetime.utcnow() # Use UTC for consistency
|
||
|
|
start_window = now - timedelta(minutes=minutes_after)
|
||
|
|
end_window = now + timedelta(minutes=minutes_ahead)
|
||
|
|
|
||
|
|
upcoming = []
|
||
|
|
|
||
|
|
for event in self.events:
|
||
|
|
try:
|
||
|
|
# Parse event date/time
|
||
|
|
date_str = event.get('date', '')
|
||
|
|
time_str = event.get('time', '')
|
||
|
|
|
||
|
|
if not date_str or not time_str:
|
||
|
|
continue
|
||
|
|
|
||
|
|
event_dt_str = f"{date_str} {time_str}"
|
||
|
|
event_dt = datetime.strptime(event_dt_str, "%Y-%m-%d %H:%M")
|
||
|
|
|
||
|
|
# Check if within window
|
||
|
|
if start_window <= event_dt <= end_window:
|
||
|
|
delta = event_dt - now
|
||
|
|
minutes_until = int(delta.total_seconds() / 60)
|
||
|
|
|
||
|
|
event_copy = event.copy()
|
||
|
|
event_copy['minutes_until'] = minutes_until
|
||
|
|
event_copy['datetime'] = event_dt.isoformat()
|
||
|
|
upcoming.append(event_copy)
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"⚠️ Error parsing event: {event.get('event', 'unknown')} - {e}")
|
||
|
|
continue
|
||
|
|
|
||
|
|
# Sort by time
|
||
|
|
upcoming.sort(key=lambda x: x['minutes_until'])
|
||
|
|
|
||
|
|
return upcoming
|
||
|
|
|
||
|
|
|
||
|
|
def is_news_upcoming(self, minutes_ahead: int = 30, minutes_after: int = 30) -> tuple:
|
||
|
|
"""
|
||
|
|
Check if high-impact news is upcoming
|
||
|
|
|
||
|
|
Args:
|
||
|
|
minutes_ahead: Block trading if news within this many minutes ahead
|
||
|
|
minutes_after: Block trading if news was this many minutes ago
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
(is_upcoming: bool, events: List[Dict])
|
||
|
|
"""
|
||
|
|
upcoming = self.get_upcoming_news(minutes_ahead, minutes_after)
|
||
|
|
|
||
|
|
# Filter only HIGH impact
|
||
|
|
high_impact = [e for e in upcoming if e.get('impact', '').upper() == 'HIGH']
|
||
|
|
|
||
|
|
return len(high_impact) > 0, high_impact
|
||
|
|
|
||
|
|
|
||
|
|
# ==========================================
|
||
|
|
# CONVENIENCE FUNCTIONS
|
||
|
|
# ==========================================
|
||
|
|
|
||
|
|
_filter = None
|
||
|
|
|
||
|
|
def get_filter() -> SimpleNewsFilter:
|
||
|
|
"""Get or create global filter instance"""
|
||
|
|
global _filter
|
||
|
|
if _filter is None:
|
||
|
|
_filter = SimpleNewsFilter()
|
||
|
|
return _filter
|
||
|
|
|
||
|
|
|
||
|
|
def is_news_upcoming(minutes_ahead: int = 30, minutes_after: int = 30) -> bool:
|
||
|
|
"""
|
||
|
|
Quick check if news is upcoming
|
||
|
|
|
||
|
|
Args:
|
||
|
|
minutes_ahead: Block if news within this many minutes
|
||
|
|
minutes_after: Block if news was this many minutes ago
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if news is upcoming, False otherwise
|
||
|
|
"""
|
||
|
|
news_filter = get_filter()
|
||
|
|
is_upcoming, events = news_filter.is_news_upcoming(minutes_ahead, minutes_after)
|
||
|
|
|
||
|
|
if is_upcoming:
|
||
|
|
print("⏸️ HIGH-IMPACT NEWS DETECTED:")
|
||
|
|
for event in events:
|
||
|
|
minutes = event['minutes_until']
|
||
|
|
time_str = f"in {minutes}min" if minutes > 0 else f"{abs(minutes)}min ago"
|
||
|
|
print(f" • {event['currency']} {event['event']} - {time_str}")
|
||
|
|
|
||
|
|
return is_upcoming
|
||
|
|
|
||
|
|
|
||
|
|
def get_upcoming_news(minutes_ahead: int = 60) -> List[Dict]:
|
||
|
|
"""
|
||
|
|
Get list of upcoming news events
|
||
|
|
|
||
|
|
Args:
|
||
|
|
minutes_ahead: Look ahead this many minutes
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
List of news events
|
||
|
|
"""
|
||
|
|
news_filter = get_filter()
|
||
|
|
return news_filter.get_upcoming_news(minutes_ahead)
|
||
|
|
|
||
|
|
|
||
|
|
def reload_events():
|
||
|
|
"""Reload events from config file"""
|
||
|
|
global _filter
|
||
|
|
_filter = None
|
||
|
|
return get_filter()
|
||
|
|
|
||
|
|
|
||
|
|
# ==========================================
|
||
|
|
# TESTING
|
||
|
|
# ==========================================
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
print("="*70)
|
||
|
|
print("📰 NEWS FILTER SIMPLE - TEST")
|
||
|
|
print("="*70)
|
||
|
|
print()
|
||
|
|
|
||
|
|
# Initialize filter
|
||
|
|
news_filter = SimpleNewsFilter()
|
||
|
|
|
||
|
|
# Test 1: Show all loaded events
|
||
|
|
print("Test 1: Loaded Events")
|
||
|
|
print(f"Total events: {len(news_filter.events)}")
|
||
|
|
for i, event in enumerate(news_filter.events, 1):
|
||
|
|
print(f" {i}. [{event.get('currency')}] {event.get('event')}")
|
||
|
|
print(f" Date: {event.get('date')} {event.get('time')} {event.get('timezone')}")
|
||
|
|
print()
|
||
|
|
|
||
|
|
# Test 2: Upcoming events (next 24 hours)
|
||
|
|
print("Test 2: Upcoming events (next 24 hours)")
|
||
|
|
upcoming = news_filter.get_upcoming_news(minutes_ahead=1440)
|
||
|
|
|
||
|
|
if upcoming:
|
||
|
|
print(f"Found {len(upcoming)} upcoming events:")
|
||
|
|
for event in upcoming:
|
||
|
|
minutes = event['minutes_until']
|
||
|
|
time_str = f"in {minutes}min" if minutes > 0 else f"{abs(minutes)}min ago"
|
||
|
|
print(f" • [{event['currency']}] {event['event']}")
|
||
|
|
print(f" {time_str} ({event.get('date')} {event.get('time')} UTC)")
|
||
|
|
else:
|
||
|
|
print("No upcoming events in next 24 hours")
|
||
|
|
print()
|
||
|
|
|
||
|
|
# Test 3: Should trading be blocked?
|
||
|
|
print("Test 3: Should trading be blocked? (30min window)")
|
||
|
|
is_upcoming, blocking_events = news_filter.is_news_upcoming(
|
||
|
|
minutes_ahead=30,
|
||
|
|
minutes_after=30
|
||
|
|
)
|
||
|
|
|
||
|
|
if is_upcoming:
|
||
|
|
print("❌ YES - Trading should be BLOCKED")
|
||
|
|
print("Blocking events:")
|
||
|
|
for event in blocking_events:
|
||
|
|
minutes = event['minutes_until']
|
||
|
|
time_str = f"in {minutes}min" if minutes > 0 else f"{abs(minutes)}min ago"
|
||
|
|
print(f" • [{event['currency']}] {event['event']} - {time_str}")
|
||
|
|
else:
|
||
|
|
print("✅ NO - Trading is safe")
|
||
|
|
|
||
|
|
print()
|
||
|
|
print("="*70)
|
||
|
|
print("✅ TEST COMPLETE")
|
||
|
|
print("="*70)
|
||
|
|
print()
|
||
|
|
print("📝 To add events, edit: news_events_manual.json")
|