From 6d3bcea4350629a1bf5a94827e3eee67ca715349 Mon Sep 17 00:00:00 2001 From: pie Date: Mon, 6 Jul 2026 15:54:33 +0100 Subject: [PATCH] fix: resolve NameError for pandas and overhaul logging to capture all sub-script output in real-time --- main.py | 208 ++++++++++++++++++++------------------------------------ 1 file changed, 72 insertions(+), 136 deletions(-) diff --git a/main.py b/main.py index 0d777aa..cb60389 100644 --- a/main.py +++ b/main.py @@ -11,6 +11,7 @@ import pytz import threading import csv import random +import pandas as pd from datetime import datetime, time as dtime from dotenv import load_dotenv @@ -20,51 +21,66 @@ from src.execution.manager import ExecutionManager from scripts.find_isa_candidates import find_best_isa_tickers from scripts.backtest import backtest_ticker -# Aggressive flush handler to ensure logs are physically written to disk -class HardFlushHandler(logging.handlers.WatchedFileHandler): - def emit(self, record): - super().emit(record) - # Flush the internal stream buffer - self.flush() - # Force the OS to write the bits to the physical drive - if self.stream and not self.stream.closed: - try: - os.fsync(self.stream.fileno()) - except (OSError, ValueError): - pass - # Ensure logs directory exists os.makedirs("logs", exist_ok=True) log_filename = datetime.now().strftime("logs/bot_%Y-%m-%d.log") -# Robust logging setup +# Save original stdout/stderr to avoid recursion loops when redirecting +_original_stdout = sys.stdout +_original_stderr = sys.stderr + +# Aggressive flush handler to ensure logs are physically written to disk +class HardFlushHandler(logging.FileHandler): + def emit(self, record): + super().emit(record) + self.flush() + if self.stream and not self.stream.closed and hasattr(self.stream, 'fileno'): + try: + os.fsync(self.stream.fileno()) + except: + pass + +# Configure logging +# We log to the file using our HardFlushHandler and to the ORIGINAL stdout for systemd +file_handler = HardFlushHandler(log_filename, mode='a') +stream_handler = logging.StreamHandler(_original_stdout) + logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(threadName)s] %(levelname)s - %(message)s', - handlers=[ - HardFlushHandler(log_filename, mode='a'), - logging.StreamHandler(sys.stdout) - ] + handlers=[file_handler, stream_handler] ) logger = logging.getLogger(__name__) +# Redirect all print() statements to the logger +class LoggerWriter: + def __init__(self, level): + self.level = level + def write(self, message): + if message.strip(): + # This logs to BOTH the file and the ORIGINAL stdout + logger.log(self.level, message.strip()) + def flush(self): + for handler in logger.handlers: + handler.flush() + +sys.stdout = LoggerWriter(logging.INFO) +sys.stderr = LoggerWriter(logging.ERROR) + def flush_logs(): for handler in logging.getLogger().handlers: handler.flush() - if hasattr(handler, 'stream') and handler.stream and not handler.stream.closed: + if hasattr(handler, 'stream') and handler.stream and hasattr(handler.stream, 'fileno'): try: os.fsync(handler.stream.fileno()) except: pass - PNL_FILE = "pnl_tracking.csv" def record_pnl(ticker, direction, entry_price, exit_price, reason, pnl_r, trading_ticker=None): """Appends the result of a closed trade to the PnL CSV.""" file_exists = os.path.isfile(PNL_FILE) - - # Safety: Fix potential 0.0 exit price in logs causing extreme PnL values if exit_price <= 0: exit_price = entry_price @@ -72,7 +88,6 @@ def record_pnl(ticker, direction, entry_price, exit_price, reason, pnl_r, tradin writer = csv.writer(file) if not file_exists: writer.writerow(["Date", "Ticker", "Trading Ticker", "Direction", "Entry Price", "Exit Price", "Reason", "PnL (R)"]) - today = datetime.now().strftime("%Y-%m-%d %H:%M:%S") writer.writerow([today, ticker, trading_ticker or ticker, direction, round(entry_price, 2), round(exit_price, 2), reason, round(pnl_r, 2)]) @@ -82,10 +97,8 @@ def record_pnl(ticker, direction, entry_price, exit_price, reason, pnl_r, tradin def calculate_r_multiple(direction, entry_price, exit_price, stop_loss): """Calculates the PnL in terms of Risk Multiples (R).""" - # Safety: Prevent Division by Zero if SL is somehow same as entry if abs(entry_price - stop_loss) < 0.001: return 0.0 - if direction == "BUY": # LONG risk = entry_price - stop_loss return (exit_price - entry_price) / risk if risk != 0 else 0 @@ -93,9 +106,10 @@ def calculate_r_multiple(direction, entry_price, exit_price, stop_loss): risk = stop_loss - entry_price return (entry_price - exit_price) / risk if risk != 0 else 0 -def run_ticker_lifecycle(isa_client, cfd_client, yf_ticker, t212_ticker, tz): +def run_ticker_lifecycle(client, yf_ticker, t212_ticker, tz): """Handles the full strategy lifecycle for a single ticker.""" strategy = TouchTurnStrategy(yf_ticker) + execution = ExecutionManager(client) logger.info(f"Bot thread started for {yf_ticker} ({t212_ticker}).") @@ -103,110 +117,76 @@ def run_ticker_lifecycle(isa_client, cfd_client, yf_ticker, t212_ticker, tz): now = datetime.now(tz) target_entry_time = now.replace(hour=9, minute=45, second=0, microsecond=0) - # 1. Wait until 09:45 EST if now < target_entry_time: wait_seconds = (target_entry_time - now).total_seconds() logger.info(f"Waiting {wait_seconds:.0f} seconds until 09:45 EST evaluation...") time.sleep(wait_seconds) - # Re-evaluate current time now = datetime.now(tz) - if now.hour == 9 and now.minute >= 45: logger.info(f"Evaluating opening candle for {yf_ticker}...") - # Retry loop: wait for yfinance to publish the 09:30-09:45 candle setup_found = False - max_retries = 12 - for attempt in range(max_retries): + for attempt in range(12): if strategy.check_setup(): setup_found = True break - elif attempt < max_retries - 1: - logger.debug(f"Data not ready for {yf_ticker} yet, waiting 15s...") + elif attempt < 11: time.sleep(15) if setup_found: params = strategy.get_trade_params() params['ticker'] = t212_ticker - # Split-Account Routing Logic - split_mode = os.getenv("SPLIT_ACCOUNT_MODE", "False").lower() == "true" - isa_mode = os.getenv("ISA_MODE", "False").lower() == "true" - - client = isa_client - use_isa_rules = True - - if params['direction'] == "SELL": - if split_mode and cfd_client: - logger.info(f"Split-Account Mode: Routing SHORT trade for {yf_ticker} to CFD account.") - client = cfd_client - use_isa_rules = False # Disable Inverse ETP mapping for CFD - elif isa_mode: - # Standard ISA mode check - from src.strategy.inverse_mapping import INVERSE_TICKER_MAP - base_ticker = yf_ticker.split('_')[0] - if base_ticker not in INVERSE_TICKER_MAP: - logger.warning(f"ISA Mode: Bypassing {yf_ticker} Short (No ETP).") - return - - execution = ExecutionManager(client) + from src.strategy.inverse_mapping import INVERSE_TICKER_MAP + if params['direction'] == "SELL" and yf_ticker.split('_')[0] not in INVERSE_TICKER_MAP: + logger.warning(f"ISA Mode: Bypassing {yf_ticker} Short (No ETP).") + return - # Anti-thundering-herd jitter time.sleep(random.uniform(1.0, 10.0)) - # Fetch Account Balance and DYNAMICALLY partition for attempt in range(3): try: account_info = client.get_account_info() actual_balance = float(account_info.get('totalValue', 5000.0)) virtual_balance = max(0, actual_balance - 4750.0) - # Count actively trading threads on THIS account num_active = 0 for t in threading.enumerate(): if t.name.startswith("Bot-") and t.is_alive(): num_active += 1 - num_active = max(1, num_active) # Safety + num_active = max(1, num_active) risk_share = (virtual_balance * 0.05) / num_active capital_share = virtual_balance / num_active - logger.info(f"Account ({'ISA' if use_isa_rules else 'CFD'}): Active Trades: {num_active} | Virtual: {virtual_balance:.2f} | Risk: {risk_share:.2f}") + logger.info(f"Active Trades: {num_active} | Virtual: {virtual_balance:.2f} | Share: {capital_share:.2f} | Risk: {risk_share:.2f}") break except Exception as e: if '429' in str(e): - wait_time = (attempt + 1) * 5 + random.uniform(1, 3) - logger.warning(f"Rate limited on account fetch for {yf_ticker}. Retrying in {wait_time:.1f}s...") - time.sleep(wait_time) + time.sleep((attempt + 1) * 5) else: - logger.error(f"Failed to fetch account info: {e}") break - if execution.execute_trade(params, target_risk_amount=risk_share, max_capital=capital_share, isa_rules=use_isa_rules): + if execution.execute_trade(params, target_risk_amount=risk_share, max_capital=capital_share, isa_rules=True): if execution.monitor_and_bracket(params): - # Position is open, monitor for exit via SL/TP while datetime.now(tz).hour < 11: is_closed, reason, exit_price = execution.check_exit_status() if is_closed: final_entry = execution.params.get('final_entry', params['entry_price']) final_sl = execution.params.get('final_sl', params['stop_loss']) trading_ticker = execution.params.get('trading_ticker', yf_ticker) - pnl_r = calculate_r_multiple("BUY" if execution.is_etp else params['direction'], final_entry, exit_price, final_sl) record_pnl(yf_ticker, params['direction'], final_entry, exit_price, reason, pnl_r, trading_ticker=trading_ticker) break - time.sleep(15) now = datetime.now(tz) else: logger.info(f"No valid setup today for {yf_ticker}. Thread exiting.") return - # 2. Wait until 11:00 EST for Forced Exit now = datetime.now(tz) target_exit_time = now.replace(hour=11, minute=0, second=0, microsecond=0) - if now < target_exit_time and execution.is_in_position: wait_seconds = (target_exit_time - now).total_seconds() logger.info(f"Waiting {wait_seconds:.0f} seconds until 11:00 EST forced exit...") @@ -215,20 +195,17 @@ def run_ticker_lifecycle(isa_client, cfd_client, yf_ticker, t212_ticker, tz): except Exception as e: logger.error(f"Unexpected error in {yf_ticker} lifecycle: {e}", exc_info=True) finally: - # 3. 11:00 EST - Cleanup (ensures closing even on thread crash) time.sleep(random.uniform(0.1, 5.0)) - logger.info(f"Cleanup phase reached for {yf_ticker}.") - if execution and execution.is_in_position: + if execution.is_in_position: exit_price = execution.close_all(t212_ticker) if hasattr(execution, 'params') and exit_price > 0: final_entry = execution.params.get('final_entry', execution.params['entry_price']) final_sl = execution.params.get('final_sl', execution.params['stop_loss']) trading_ticker = execution.params.get('trading_ticker', yf_ticker) - pnl_r = calculate_r_multiple("BUY" if execution.is_etp else execution.params['direction'], final_entry, exit_price, final_sl) record_pnl(yf_ticker, execution.params['direction'], final_entry, exit_price, "Forced Exit (Final)", pnl_r, trading_ticker=trading_ticker) - elif execution: + else: execution.close_all(t212_ticker) logger.info(f"Lifecycle complete for {yf_ticker}. Thread exiting.") @@ -236,52 +213,33 @@ def run_ticker_lifecycle(isa_client, cfd_client, yf_ticker, t212_ticker, tz): def main(): load_dotenv() + logger.info("Touch & Turn Bot Initializing...") - # Primary Account (ISA) - isa_key_id = os.getenv("TRADING212_API_KEY_ID") - isa_key = os.getenv("TRADING212_API_KEY") - isa_url = os.getenv("TRADING212_BASE_URL", "https://demo.trading212.com/api/v0/") - - # Secondary Account (CFD) - cfd_key_id = os.getenv("CFD_API_KEY_ID") - cfd_key = os.getenv("CFD_API_KEY") - cfd_url = os.getenv("CFD_BASE_URL", "https://demo.trading212.com/api/v0/") - + key_id = os.getenv("TRADING212_API_KEY_ID") + key = os.getenv("TRADING212_API_KEY") + url = os.getenv("TRADING212_BASE_URL", "https://demo.trading212.com/api/v0/") tz = pytz.timezone('US/Eastern') - now = datetime.now(tz) - if now.weekday() >= 5: - logger.warning("Weekend detected. The market is closed. Exiting cleanly.") + if datetime.now(tz).weekday() >= 5: + logger.warning("Weekend detected. Exiting cleanly.") return - if now.hour < 9 or (now.hour == 9 and now.minute > 40) or now.hour >= 10: - logger.warning(f"Bot executed at {now.strftime('%H:%M')} EST. Expected launch window is 09:00 - 09:40 EST. Exiting cleanly.") + if not key_id or not key: + logger.error("API credentials not found in .env") return - if not isa_key_id or not isa_key: - logger.error("Primary API credentials not found in .env") - return + client = Trading212Client(key_id, key, url) - isa_client = Trading212Client(isa_key_id, isa_key, isa_url) - cfd_client = None - if cfd_key_id and cfd_key: - cfd_client = Trading212Client(cfd_key_id, cfd_key, cfd_url) - - # Early verification try: - logger.info("Verifying Primary API connection...") - isa_client.get_account_info() - if cfd_client: - logger.info("Verifying Secondary API connection...") - cfd_client.get_account_info() - logger.info("API Connections verified successfully.") + logger.info("Verifying API connection...") + client.get_account_info() + logger.info("API Connection verified successfully.") except Exception as e: logger.error(f"API Connection check failed: {e}") return logger.info("Starting Morning Routine: Finding ISA Candidates...") candidates_df = find_best_isa_tickers() - if candidates_df is None or candidates_df.empty: logger.error("No candidates found. Exiting.") return @@ -289,39 +247,25 @@ def main(): logger.info("Running Backtests on top 10 candidates to find the current 'Edge'...") all_results = [] profitable_tickers = [] - for _, row in candidates_df.head(10).iterrows(): yf_t = row['Ticker'] - t212_t = row['T212_Ticker'] - res = backtest_ticker(yf_t, quiet=True) if res: all_results.append(res) if res['Net PnL (R)'] > 0: - profitable_tickers.append({ - 'yf': yf_t, - 't212': t212_t, - 'pnl': res['Net PnL (R)'] - }) + profitable_tickers.append({'yf': yf_t, 't212': row['T212_Ticker'], 'pnl': res['Net PnL (R)']}) if all_results: from prettytable import PrettyTable - results_df = pd.DataFrame(all_results) - results_df = results_df.sort_values(by="Net PnL (R)", ascending=False).reset_index(drop=True) - - print("\n" + "="*80) - print("šŸš€ MORNING BACKTEST LEADERBOARD (LAST ~60 DAYS) šŸš€") - print("="*80) + results_df = pd.DataFrame(all_results).sort_values(by="Net PnL (R)", ascending=False).reset_index(drop=True) + print("\n" + "="*80 + "\nšŸš€ MORNING BACKTEST LEADERBOARD (LAST ~60 DAYS) šŸš€\n" + "="*80) table = PrettyTable() table.field_names = results_df.columns - for _, r in results_df.iterrows(): - table.add_row(r.tolist()) + for _, r in results_df.iterrows(): table.add_row(r.tolist()) print(table) - print("\n") profitable_tickers.sort(key=lambda x: x['pnl'], reverse=True) final_watchlist = profitable_tickers[:3] - if not final_watchlist: logger.warning("No tickers showed a positive backtest return. Bot will not trade today.") return @@ -330,19 +274,11 @@ def main(): threads = [] for ticker_info in final_watchlist: - t = threading.Thread( - target=run_ticker_lifecycle, - args=(isa_client, cfd_client, ticker_info['yf'], ticker_info['t212'], tz), - name=f"Bot-{ticker_info['yf']}" - ) + t = threading.Thread(target=run_ticker_lifecycle, args=(client, ticker_info['yf'], ticker_info['t212'], tz), name=f"Bot-{ticker_info['yf']}") t.start() threads.append(t) - logger.info("All execution threads launched. Waiting for completion...") - - for t in threads: - t.join() - + for t in threads: t.join() logger.info("All threads completed. Bot shutting down for the day.") flush_logs() @@ -350,7 +286,7 @@ if __name__ == "__main__": try: main() except Exception as e: - logger.critical(f"FATAL ERROR in main: {e}", exc_info=True) + # Avoid recursion by printing to original stderr + print(f"FATAL ERROR in main: {e}", file=_original_stderr) finally: flush_logs() - logger.info("Bot process terminated.")