Place-Order-Trading-Bot/news_filter_simple.py

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