diff --git a/equity_curve_trading.py b/equity_curve_trading.py index 0560090..1a95f80 100644 --- a/equity_curve_trading.py +++ b/equity_curve_trading.py @@ -24,7 +24,7 @@ VERWENDUNG: import json import os -from datetime import datetime +from datetime import datetime, timezone, timedelta from typing import List, Dict, Optional, Tuple import logging @@ -111,7 +111,8 @@ class EquityCurveManager: # Aktuelle Equity holen current_equity = self._get_current_equity(mt5_account_info) if current_equity is None: - return True, "Could not get equity, allowing trade", 1.0 + logger.error("Could not get equity from MT5 — blocking trade as safe default") + return False, "Equity unavailable — trade blocked for safety", 0.0 # MA berechnen ma_equity = self._calculate_ma() @@ -157,7 +158,7 @@ class EquityCurveManager: return entry = { - "timestamp": datetime.now().isoformat(), + "timestamp": datetime.now(timezone.utc).isoformat(), "equity": current_equity, "trade_count": len(self.equity_history) + 1 } @@ -255,8 +256,8 @@ class EquityCurveManager: return mt5_account_info.equity try: - import MetaTrader5 as mt - account = mt.account_info() + import MetaTrader5 as mt5 + account = mt5.account_info() if account: return account.equity except Exception as e: @@ -306,9 +307,11 @@ class EquityCurveManager: Nützlich wenn du mit bestehendem Konto startest """ + base_time = datetime.now(timezone.utc) for i in range(self.min_trades): + ts = (base_time - timedelta(minutes=self.min_trades - i)).isoformat() self.equity_history.append({ - "timestamp": datetime.now().isoformat(), + "timestamp": ts, "equity": equity, "trade_count": i + 1, "note": "Initial warmup entry" diff --git a/infrastructure_patch.py b/infrastructure_patch.py index cc39554..ecfc48c 100644 --- a/infrastructure_patch.py +++ b/infrastructure_patch.py @@ -16,10 +16,13 @@ USAGE: from trading_database import TradingDatabase from telegram_notifier import TelegramNotifier, load_telegram_config -from datetime import datetime +from datetime import datetime, timezone from typing import Dict, Optional, Callable import json import os +import logging + +logger = logging.getLogger(__name__) class TradingInfrastructure: @@ -50,9 +53,9 @@ class TradingInfrastructure: if self.enable_database: try: self.db = TradingDatabase(db_path) - print(f"✅ Database initialized: {db_path}") + logger.info(f"Database initialized: {db_path}") except Exception as e: - print(f"❌ Database initialization failed: {e}") + logger.error(f"Database initialization failed: {e}") self.enable_database = False self.db = None else: @@ -69,13 +72,13 @@ class TradingInfrastructure: bot_token=telegram_config['bot_token'], chat_id=telegram_config['chat_id'] ) - print("✅ Telegram notifications enabled") + logger.info("Telegram notifications enabled") else: - print("⚠️ No Telegram config found, notifications disabled") + logger.warning("No Telegram config found, notifications disabled") self.enable_telegram = False self.telegram = None except Exception as e: - print(f"❌ Telegram initialization failed: {e}") + logger.error(f"Telegram initialization failed: {e}") self.enable_telegram = False self.telegram = None else: @@ -94,18 +97,17 @@ class TradingInfrastructure: trade_data: Dictionary with trade information """ # Log to database - if self.enable_database and self.db: + if self.db: try: self.db.log_trade_entry(trade_data) except Exception as e: - print(f"⚠️ Database logging failed: {e}") + logger.error(f"Database logging failed: {e}") - # Send Telegram notification - if self.enable_telegram and self.telegram: + if self.telegram: try: self.telegram.notify_trade_entry(trade_data) except Exception as e: - print(f"⚠️ Telegram notification failed: {e}") + logger.error(f"Telegram notification failed: {e}") def log_trade_exit(self, ticket: int, exit_data: Dict): @@ -117,32 +119,28 @@ class TradingInfrastructure: exit_data: Dictionary with exit information """ # Update database - if self.enable_database and self.db: + if self.db: try: self.db.update_trade_exit(ticket, exit_data) except Exception as e: - print(f"⚠️ Database update failed: {e}") + logger.error(f"Database update failed: {e}") # Send Telegram notification - if self.enable_telegram and self.telegram: + if self.telegram: try: - # Get full trade data from database if self.db: - trades = self.db.cursor.execute( - "SELECT * FROM trades WHERE ticket = ?", - (ticket,) - ).fetchone() - - if trades: - trade_dict = dict(trades) + open_trades = self.db.get_open_trades() + trade_row = next((t for t in open_trades if t.get('ticket') == ticket), None) + if trade_row: + trade_dict = dict(trade_row) trade_dict.update(exit_data) self.telegram.notify_trade_exit(trade_dict) + else: + self.telegram.notify_trade_exit(exit_data) else: - # Fallback if no database self.telegram.notify_trade_exit(exit_data) - except Exception as e: - print(f"⚠️ Telegram notification failed: {e}") + logger.error(f"Telegram notification failed: {e}") # ========================================== @@ -201,6 +199,15 @@ class TradingInfrastructure: Returns: Dictionary with trade data """ + entry_time = datetime.fromtimestamp(position.time, tz=timezone.utc).replace(tzinfo=None) + + # risk_pct: try to read from SESSION_WHITELIST_CONFIG, fall back to 0.01 + try: + from session_filter_patch import SESSION_WHITELIST_CONFIG + risk_pct = SESSION_WHITELIST_CONFIG.get('max_risk_per_trade', 0.01) + except Exception: + risk_pct = 0.01 + return { 'ticket': position.ticket, 'position_id': position.identifier, @@ -211,11 +218,11 @@ class TradingInfrastructure: 'entry_price': position.price_open, 'sl_price': position.sl, 'tp_price': position.tp, - 'entry_time': datetime.fromtimestamp(position.time).strftime('%Y-%m-%d %H:%M:%S'), + 'entry_time': entry_time.strftime('%Y-%m-%d %H:%M:%S'), 'session': session, 'confidence': confidence, 'quality': quality, - 'risk_pct': 0.01, # From config + 'risk_pct': risk_pct, 'status': 'open' } @@ -231,11 +238,11 @@ class TradingInfrastructure: Args: config: Bot configuration """ - if self.enable_telegram and self.telegram: + if self.telegram: try: self.telegram.send_bot_started(config) except Exception as e: - print(f"⚠️ Telegram notification failed: {e}") + logger.error(f"Telegram notification failed: {e}") def send_bot_stopped(self, reason: str = "Manual stop"): @@ -245,28 +252,28 @@ class TradingInfrastructure: Args: reason: Stop reason """ - if self.enable_telegram and self.telegram: + if self.telegram: try: self.telegram.send_bot_stopped(reason) except Exception as e: - print(f"⚠️ Telegram notification failed: {e}") + logger.error(f"Telegram notification failed: {e}") def send_daily_report(self): """Send daily performance report""" - if not (self.enable_database and self.enable_telegram): + if not (self.db and self.telegram): return try: stats = self.db.get_daily_summary() self.telegram.send_daily_report(stats) except Exception as e: - print(f"⚠️ Daily report failed: {e}") + logger.error(f"Daily report failed: {e}") def send_weekly_report(self): """Send weekly performance report""" - if not (self.enable_database and self.enable_telegram): + if not (self.db and self.telegram): return try: @@ -274,7 +281,7 @@ class TradingInfrastructure: session_perf = self.db.get_session_performance(days=7) self.telegram.send_weekly_report(stats, session_perf) except Exception as e: - print(f"⚠️ Weekly report failed: {e}") + logger.error(f"Weekly report failed: {e}") def log_bot_status(self, status: str, config: Dict = None, error_message: str = None): @@ -286,7 +293,7 @@ class TradingInfrastructure: config: Optional bot configuration error_message: Optional error message """ - if not (self.enable_database and self.db): + if not self.db: return try: @@ -301,7 +308,7 @@ class TradingInfrastructure: self.db.log_bot_status(status_data) except Exception as e: - print(f"⚠️ Status logging failed: {e}") + logger.error(f"Status logging failed: {e}") # ========================================== @@ -349,7 +356,7 @@ class TradingInfrastructure: Returns: Dictionary with statistics """ - if not (self.enable_database and self.db): + if not self.db: return {} try: @@ -361,7 +368,7 @@ class TradingInfrastructure: 'sessions': session_perf } except Exception as e: - print(f"⚠️ Performance summary failed: {e}") + logger.error(f"Performance summary failed: {e}") return {} diff --git a/news_filter.py b/news_filter.py deleted file mode 100644 index 27e2e89..0000000 --- a/news_filter.py +++ /dev/null @@ -1,378 +0,0 @@ -#!/usr/bin/env python3 -""" -📰 NEWS FILTER - Economic Calendar Integration -Blockiert Trading vor/nach High-Impact News Events - -FEATURES: -- ForexFactory Calendar Scraping (kostenlos, keine API Key nötig) -- High-Impact Events Filter -- 30min Buffer vor/nach Event -- USD, EUR, GBP News (relevant für Gold) -- Cache für Performance - -VERWENDUNG: - from news_filter import is_news_upcoming, get_upcoming_news - - if is_news_upcoming(minutes_ahead=30): - print("⏸️ Trading SKIP: High-Impact News Event") - return -""" - -import requests -from datetime import datetime, timedelta -from bs4 import BeautifulSoup -import json -import os -from typing import List, Dict, Optional - - -class EconomicCalendar: - """ - Economic Calendar - ForexFactory Scraper - """ - - def __init__(self, cache_minutes: int = 15): - """ - Initialize Economic Calendar - - Args: - cache_minutes: Cache duration in minutes - """ - self.cache_file = "news_cache.json" - self.cache_duration = timedelta(minutes=cache_minutes) - self.last_fetch = None - self.cached_events = [] - - - def _get_forexfactory_calendar(self) -> List[Dict]: - """ - Scrape ForexFactory Calendar - - Returns: - List of news events - """ - try: - # ForexFactory calendar URL - url = "https://www.forexfactory.com/calendar" - - headers = { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' - } - - response = requests.get(url, headers=headers, timeout=10) - - if response.status_code != 200: - print(f"⚠️ ForexFactory returned status {response.status_code}") - return [] - - soup = BeautifulSoup(response.content, 'html.parser') - - # Parse calendar events - events = [] - calendar_rows = soup.find_all('tr', class_='calendar__row') - - current_date = datetime.now().date() - - for row in calendar_rows: - try: - # Get time - time_elem = row.find('td', class_='calendar__time') - if not time_elem: - continue - - time_text = time_elem.get_text(strip=True) - if not time_text or time_text in ['All Day', 'Day']: - continue - - # Get currency - currency_elem = row.find('td', class_='calendar__currency') - if not currency_elem: - continue - currency = currency_elem.get_text(strip=True) - - # Get impact - impact_elem = row.find('td', class_='calendar__impact') - if not impact_elem: - continue - - impact_spans = impact_elem.find_all('span', class_='icon--ff-impact-gra') - impact_level = len(impact_spans) # 1=low, 2=medium, 3=high - - # Get event name - event_elem = row.find('td', class_='calendar__event') - if not event_elem: - continue - event_name = event_elem.get_text(strip=True) - - # Only track USD, EUR, GBP (relevant for Gold) - if currency not in ['USD', 'EUR', 'GBP']: - continue - - # Only track high-impact events (level 3) - if impact_level < 3: - continue - - # Parse time - try: - event_time = datetime.strptime( - f"{current_date} {time_text}", - "%Y-%m-%d %I:%M%p" - ) - except: - continue - - events.append({ - 'time': event_time.isoformat(), - 'currency': currency, - 'event': event_name, - 'impact': 'HIGH', - 'impact_level': impact_level - }) - - except Exception as e: - # Skip malformed rows - continue - - return events - - except Exception as e: - print(f"❌ Error fetching ForexFactory calendar: {e}") - return [] - - - def _load_cache(self) -> bool: - """Load events from cache""" - if not os.path.exists(self.cache_file): - return False - - try: - with open(self.cache_file, 'r') as f: - data = json.load(f) - - cache_time = datetime.fromisoformat(data['timestamp']) - - # Check if cache is still valid - if datetime.now() - cache_time < self.cache_duration: - self.cached_events = data['events'] - self.last_fetch = cache_time - return True - - except Exception as e: - print(f"⚠️ Cache load error: {e}") - - return False - - - def _save_cache(self, events: List[Dict]): - """Save events to cache""" - try: - data = { - 'timestamp': datetime.now().isoformat(), - 'events': events - } - - with open(self.cache_file, 'w') as f: - json.dump(data, f, indent=2) - - except Exception as e: - print(f"⚠️ Cache save error: {e}") - - - def get_news_events(self, force_refresh: bool = False) -> List[Dict]: - """ - Get news events (cached or fresh) - - Args: - force_refresh: Force fetch from API - - Returns: - List of news events - """ - # Try cache first - if not force_refresh and self._load_cache(): - return self.cached_events - - # Fetch fresh data - print("📰 Fetching economic calendar...") - events = self._get_forexfactory_calendar() - - if events: - self.cached_events = events - self.last_fetch = datetime.now() - self._save_cache(events) - print(f"✅ Fetched {len(events)} high-impact events") - else: - print("⚠️ No events fetched, using cache if available") - self._load_cache() - - return self.cached_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 - """ - events = self.get_news_events() - - now = datetime.now() - start_window = now - timedelta(minutes=minutes_after) - end_window = now + timedelta(minutes=minutes_ahead) - - upcoming = [] - - for event in events: - try: - event_time = datetime.fromisoformat(event['time']) - - if start_window <= event_time <= end_window: - # Add time until event - delta = event_time - now - minutes_until = int(delta.total_seconds() / 60) - - event_copy = event.copy() - event_copy['minutes_until'] = minutes_until - upcoming.append(event_copy) - - except Exception as 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) - - return len(upcoming) > 0, upcoming - - -# ========================================== -# CONVENIENCE FUNCTIONS -# ========================================== - -# Global calendar instance -_calendar = None - -def get_calendar() -> EconomicCalendar: - """Get or create global calendar instance""" - global _calendar - if _calendar is None: - _calendar = EconomicCalendar() - return _calendar - - -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 - """ - calendar = get_calendar() - is_upcoming, events = calendar.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 - """ - calendar = get_calendar() - return calendar.get_upcoming_news(minutes_ahead) - - -def force_refresh_calendar(): - """Force refresh calendar data""" - calendar = get_calendar() - calendar.get_news_events(force_refresh=True) - - -# ========================================== -# TESTING -# ========================================== - -if __name__ == "__main__": - print("="*70) - print("📰 NEWS FILTER - TEST") - print("="*70) - print() - - # Test 1: Fetch calendar - print("Test 1: Fetching calendar...") - calendar = EconomicCalendar() - events = calendar.get_news_events(force_refresh=True) - print(f"✅ Found {len(events)} high-impact events") - print() - - # Test 2: Show upcoming events - print("Test 2: Upcoming events (next 4 hours)...") - upcoming = calendar.get_upcoming_news(minutes_ahead=240) - - 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']} - {time_str}") - else: - print("No upcoming high-impact events in next 4 hours") - print() - - # Test 3: Check if trading should be blocked - print("Test 3: Should trading be blocked? (30min window)") - is_upcoming, blocking_events = calendar.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) diff --git a/news_filter_v2.py b/news_filter_v2.py deleted file mode 100644 index ff9f77e..0000000 --- a/news_filter_v2.py +++ /dev/null @@ -1,445 +0,0 @@ -#!/usr/bin/env python3 -""" -📰 NEWS FILTER V2 - Economic Calendar (Finnhub API) -Blockiert Trading vor/nach High-Impact News Events - -FREE API: Finnhub (60 calls/minute free tier) -Sign up: https://finnhub.io/register - -FEATURES: -- Finnhub Economic Calendar API -- High-Impact Events Filter -- 30min Buffer vor/nach Event -- USD, EUR, GBP News (relevant für Gold) -- Cache für Performance - -SETUP: -1. Sign up at https://finnhub.io/register -2. Get your free API key -3. Add to news_config.json: - {"finnhub_api_key": "YOUR_API_KEY_HERE"} - -VERWENDUNG: - from news_filter_v2 import is_news_upcoming, get_upcoming_news - - if is_news_upcoming(minutes_ahead=30): - print("⏸️ Trading SKIP: High-Impact News Event") - return -""" - -import requests -from datetime import datetime, timedelta -import json -import os -from typing import List, Dict, Optional - - -class EconomicCalendarV2: - """ - Economic Calendar V2 - Finnhub API - """ - - def __init__(self, api_key: Optional[str] = None, cache_minutes: int = 15): - """ - Initialize Economic Calendar - - Args: - api_key: Finnhub API key (or load from config) - cache_minutes: Cache duration in minutes - """ - # Load API key - if api_key: - self.api_key = api_key - else: - self.api_key = self._load_api_key() - - self.cache_file = "news_cache.json" - self.cache_duration = timedelta(minutes=cache_minutes) - self.last_fetch = None - self.cached_events = [] - - # High-impact event keywords - self.high_impact_keywords = [ - 'NFP', 'Non-Farm', 'Payrolls', 'Employment', - 'CPI', 'Inflation', 'Consumer Price', - 'GDP', 'Gross Domestic Product', - 'FOMC', 'Fed', 'Federal Reserve', 'Interest Rate', - 'PMI', 'Manufacturing', - 'Retail Sales', - 'Unemployment', - 'ECB', 'European Central Bank', - 'BOE', 'Bank of England' - ] - - - def _load_api_key(self) -> str: - """Load API key from config file""" - config_file = "news_config.json" - - if os.path.exists(config_file): - try: - with open(config_file, 'r') as f: - config = json.load(f) - return config.get('finnhub_api_key', '') - except: - pass - - # Default demo key (limited, should replace with own) - return "demo" # Replace with your own key! - - - def _is_high_impact(self, event_name: str) -> bool: - """Check if event is high impact based on keywords""" - event_lower = event_name.lower() - - for keyword in self.high_impact_keywords: - if keyword.lower() in event_lower: - return True - - return False - - - def _get_finnhub_calendar(self) -> List[Dict]: - """ - Fetch economic calendar from Finnhub - - Returns: - List of news events - """ - try: - # Get today's date range - today = datetime.now() - from_date = (today - timedelta(days=1)).strftime('%Y-%m-%d') - to_date = (today + timedelta(days=7)).strftime('%Y-%m-%d') - - url = "https://finnhub.io/api/v1/calendar/economic" - params = { - 'token': self.api_key, - 'from': from_date, - 'to': to_date - } - - response = requests.get(url, params=params, timeout=10) - - if response.status_code != 200: - print(f"⚠️ Finnhub returned status {response.status_code}") - return [] - - data = response.json() - - if 'economicCalendar' not in data: - print("⚠️ No calendar data in response") - return [] - - # Parse events - events = [] - - for item in data['economicCalendar']: - try: - event_name = item.get('event', '') - country = item.get('country', '') - event_time_str = item.get('time', '') - - # Only USD, EUR, GBP events - if country not in ['US', 'EU', 'GB', 'UK']: - continue - - # Filter by high-impact keywords - if not self._is_high_impact(event_name): - continue - - # Parse time - try: - event_time = datetime.fromisoformat(event_time_str.replace('Z', '+00:00')) - except: - continue - - # Map country to currency - currency_map = {'US': 'USD', 'EU': 'EUR', 'GB': 'GBP', 'UK': 'GBP'} - currency = currency_map.get(country, country) - - events.append({ - 'time': event_time.isoformat(), - 'currency': currency, - 'event': event_name, - 'impact': 'HIGH', - 'country': country, - 'actual': item.get('actual'), - 'estimate': item.get('estimate'), - 'previous': item.get('previous') - }) - - except Exception as e: - continue - - return events - - except Exception as e: - print(f"❌ Error fetching Finnhub calendar: {e}") - return [] - - - def _load_cache(self) -> bool: - """Load events from cache""" - if not os.path.exists(self.cache_file): - return False - - try: - with open(self.cache_file, 'r') as f: - data = json.load(f) - - cache_time = datetime.fromisoformat(data['timestamp']) - - # Check if cache is still valid - if datetime.now() - cache_time < self.cache_duration: - self.cached_events = data['events'] - self.last_fetch = cache_time - return True - - except Exception as e: - print(f"⚠️ Cache load error: {e}") - - return False - - - def _save_cache(self, events: List[Dict]): - """Save events to cache""" - try: - data = { - 'timestamp': datetime.now().isoformat(), - 'events': events - } - - with open(self.cache_file, 'w') as f: - json.dump(data, f, indent=2) - - except Exception as e: - print(f"⚠️ Cache save error: {e}") - - - def get_news_events(self, force_refresh: bool = False) -> List[Dict]: - """ - Get news events (cached or fresh) - - Args: - force_refresh: Force fetch from API - - Returns: - List of news events - """ - # Try cache first - if not force_refresh and self._load_cache(): - return self.cached_events - - # Fetch fresh data - print("📰 Fetching economic calendar from Finnhub...") - events = self._get_finnhub_calendar() - - if events: - self.cached_events = events - self.last_fetch = datetime.now() - self._save_cache(events) - print(f"✅ Fetched {len(events)} high-impact events") - else: - print("⚠️ No events fetched, using cache if available") - self._load_cache() - - return self.cached_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 - """ - events = self.get_news_events() - - now = datetime.now() - start_window = now - timedelta(minutes=minutes_after) - end_window = now + timedelta(minutes=minutes_ahead) - - upcoming = [] - - for event in events: - try: - event_time = datetime.fromisoformat(event['time']) - - if start_window <= event_time <= end_window: - # Add time until event - delta = event_time - now - minutes_until = int(delta.total_seconds() / 60) - - event_copy = event.copy() - event_copy['minutes_until'] = minutes_until - upcoming.append(event_copy) - - except Exception as 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) - - return len(upcoming) > 0, upcoming - - -# ========================================== -# CONVENIENCE FUNCTIONS -# ========================================== - -# Global calendar instance -_calendar = None - -def get_calendar() -> EconomicCalendarV2: - """Get or create global calendar instance""" - global _calendar - if _calendar is None: - _calendar = EconomicCalendarV2() - return _calendar - - -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 - """ - calendar = get_calendar() - is_upcoming, events = calendar.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 - """ - calendar = get_calendar() - return calendar.get_upcoming_news(minutes_ahead) - - -def force_refresh_calendar(): - """Force refresh calendar data""" - calendar = get_calendar() - calendar.get_news_events(force_refresh=True) - - -# ========================================== -# SETUP HELPER -# ========================================== - -def create_config_template(): - """Create news_config.json template""" - config = { - "finnhub_api_key": "YOUR_API_KEY_HERE", - "instructions": "Get your free API key at https://finnhub.io/register" - } - - with open('news_config.json', 'w') as f: - json.dump(config, f, indent=2) - - print("✅ Created news_config.json template") - print("📝 Edit the file and add your Finnhub API key") - print(" Sign up at: https://finnhub.io/register") - - -# ========================================== -# TESTING -# ========================================== - -if __name__ == "__main__": - print("="*70) - print("📰 NEWS FILTER V2 - TEST (Finnhub API)") - print("="*70) - print() - - # Check if config exists - if not os.path.exists('news_config.json'): - print("⚠️ news_config.json not found") - print("Creating template...") - create_config_template() - print() - print("Please edit news_config.json and add your API key, then run again") - exit() - - # Test 1: Fetch calendar - print("Test 1: Fetching calendar...") - calendar = EconomicCalendarV2() - events = calendar.get_news_events(force_refresh=True) - print(f"✅ Found {len(events)} high-impact events") - print() - - # Test 2: Show upcoming events - print("Test 2: Upcoming events (next 4 hours)...") - upcoming = calendar.get_upcoming_news(minutes_ahead=240) - - 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']} - {time_str}") - else: - print("No upcoming high-impact events in next 4 hours") - print() - - # Test 3: Check if trading should be blocked - print("Test 3: Should trading be blocked? (30min window)") - is_upcoming, blocking_events = calendar.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)