#!/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")