hyper/main_app.py

import json
import logging
import os
import sys
import time
import subprocess
import multiprocessing
import schedule
import sqlite3
import pandas as pd
from datetime import datetime
import importlib
from dotenv import load_dotenv
load_dotenv()
# --- REMOVED: import signal ---
# --- REMOVED: from queue import Empty ---

from logging_utils import setup_logging
from live_market_utils import start_live_feed
from strategies.base_strategy import BaseStrategy
from dashboard import DashboardRenderer
from rich.live import Live
from hyperliquid.info import Info
from hyperliquid.utils import constants

# --- Configuration ---
WATCHED_COINS = ["BTC", "ETH", "SOL", "BNB", "HYPE", "SUI", "xyz:BRENTOIL", "xyz:CL", "xyz:GOLD", "xyz:SILVER", "mkts:USTECH", "xyz:XYZ100"]
# Display name mapping for dashboard (internal symbol -> display name)
COIN_DISPLAY_NAMES = {
    "xyz:BRENTOIL": "BRENT",
    "xyz:CL": "WTI",
    "xyz:GOLD": "GOLD",
    "xyz:SILVER": "SILVER",
    "mkts:USTECH": "USTECH",
    "xyz:XYZ100": "XYZ100"
}
LIVE_CANDLE_FETCHER_SCRIPT = "live_candle_fetcher.py"
RESAMPLER_SCRIPT = "resampler.py" 
# --- REMOVED: Market Cap Fetcher ---
# --- REMOVED: trade_executor.py is no longer a script ---
DASHBOARD_DATA_FETCHER_SCRIPT = "dashboard_data_fetcher.py"
INDICATORS_FETCHER_SCRIPT = "indicators_fetcher.py"
STRATEGY_CONFIG_FILE = os.path.join("_data", "strategies.json")
DB_PATH = os.path.join("_data", "market_data.db")
# --- REMOVED: Market Cap File ---
LOGS_DIR = "_logs"


def run_live_candle_fetcher():
    """Target function to run the live_candle_fetcher.py script in a resilient loop."""
    
    # --- GRACEFUL SHUTDOWN HANDLER ---
    import signal
    shutdown_requested = False
    
    def handle_shutdown_signal(signum, frame):
        nonlocal shutdown_requested
        # Use print here as logging may not be set up
        print(f"[CandleFetcher] Shutdown signal ({signum}) received. Will stop after current run.")
        shutdown_requested = True
            
    signal.signal(signal.SIGTERM, handle_shutdown_signal)
    signal.signal(signal.SIGINT, handle_shutdown_signal)
    # --- END GRACEFUL SHUTDOWN HANDLER ---

    log_file = os.path.join(LOGS_DIR, "live_candle_fetcher.log")
    
    while not shutdown_requested: # <-- MODIFIED
        process = None
        try:
            with open(log_file, 'a') as f:
                command = [sys.executable, LIVE_CANDLE_FETCHER_SCRIPT, "--coins"] + WATCHED_COINS + ["--log-level", "off"]
                f.write(f"\n--- Starting {LIVE_CANDLE_FETCHER_SCRIPT} at {datetime.now()} ---\n")
                
                # Use Popen instead of run to be non-blocking
                process = subprocess.Popen(command, stdout=f, stderr=subprocess.STDOUT)
                
                # Poll the process and check for shutdown request
                while process.poll() is None and not shutdown_requested:
                    time.sleep(0.5) # Poll every 500ms
                
                if shutdown_requested and process.poll() is None:
                    print(f"[CandleFetcher] Terminating subprocess {LIVE_CANDLE_FETCHER_SCRIPT}...")
                    process.terminate() # Terminate the child script
                    process.wait() # Wait for it to exit
                    print(f"[CandleFetcher] Subprocess terminated.")

        except (subprocess.CalledProcessError, Exception) as e:
            if shutdown_requested:
                break # Don't restart if we're shutting down
            with open(log_file, 'a') as f:
                f.write(f"\n--- PROCESS ERROR at {datetime.now()} ---\n")
                f.write(f"Live candle fetcher failed: {e}. Restarting...\n")
            time.sleep(5)
        
        if shutdown_requested:
            break # Exit outer loop
            
    print("[CandleFetcher] Live candle fetcher shutting down.")


def run_resampler_job(timeframes_to_generate: list):
    """Defines the job for the resampler, redirecting output to a log file."""
    log_file = os.path.join(LOGS_DIR, "resampler.log")
    try:
        command = [sys.executable, RESAMPLER_SCRIPT, "--coins"] + WATCHED_COINS + ["--timeframes"] + timeframes_to_generate + ["--log-level", "normal"]
        with open(log_file, 'a') as f:
            f.write(f"\n--- Starting resampler.py job at {datetime.now()} ---\n")
            subprocess.run(command, check=True, stdout=f, stderr=subprocess.STDOUT)
    except Exception as e:
        with open(log_file, 'a') as f:
            f.write(f"\n--- SCHEDULER ERROR at {datetime.now()} ---\n")
            f.write(f"Failed to run resampler.py job: {e}\n")


def resampler_scheduler(timeframes_to_generate: list):
    """Schedules the resampler.py script."""

    # --- GRACEFUL SHUTDOWN HANDLER ---
    import signal
    shutdown_requested = False
    
    def handle_shutdown_signal(signum, frame):
        nonlocal shutdown_requested
        try:
            logging.info(f"Shutdown signal ({signum}) received. Exiting loop...")
        except NameError:
            print(f"[ResamplerScheduler] Shutdown signal ({signum}) received. Exiting loop...")
        shutdown_requested = True
            
    signal.signal(signal.SIGTERM, handle_shutdown_signal)
    signal.signal(signal.SIGINT, handle_shutdown_signal)
    # --- END GRACEFUL SHUTDOWN HANDLER ---

    setup_logging('off', 'ResamplerScheduler')
    run_resampler_job(timeframes_to_generate)
    # Schedule to run every minute at the :01 second mark
    schedule.every().minute.at(":01").do(run_resampler_job, timeframes_to_generate=timeframes_to_generate)
    logging.info("Resampler scheduled to run every minute at :01.")
    
    while not shutdown_requested: # <-- MODIFIED
        schedule.run_pending()
        time.sleep(0.5) # Check every 500ms to not miss the scheduled time and be responsive
        
    logging.info("ResamplerScheduler shutting down.")


# --- REMOVED: run_market_cap_fetcher_job function ---

# --- REMOVED: market_cap_fetcher_scheduler function ---


def run_trade_executor(order_execution_queue: multiprocessing.Queue):
    """
    Target function to run the TradeExecutor class in a resilient loop.
    It now consumes from the order_execution_queue.
    """

    # --- GRACEFUL SHUTDOWN HANDLER ---
    import signal
    
    def handle_shutdown_signal(signum, frame):
        # We can just raise KeyboardInterrupt, as it's handled below
        logging.info(f"Shutdown signal ({signum}) received. Initiating graceful exit...")
        raise KeyboardInterrupt
            
    signal.signal(signal.SIGTERM, handle_shutdown_signal)
    # --- END GRACEFUL SHUTDOWN HANDLER ---

    log_file_path = os.path.join(LOGS_DIR, "trade_executor.log")
    try:
        sys.stdout = open(log_file_path, 'a', buffering=1)
        sys.stderr = sys.stdout
    except Exception as e:
        print(f"Failed to open log file for TradeExecutor: {e}")

    setup_logging('normal', f"TradeExecutor")
    logging.info("\n--- Starting Trade Executor process ---")
    
    while True:
        try:
            from trade_executor import TradeExecutor
            
            executor = TradeExecutor(log_level="normal", order_execution_queue=order_execution_queue)
            
            # --- REVERTED: Call executor.run() directly ---
            executor.run() 
            
        except KeyboardInterrupt:
            logging.info("Trade Executor interrupted. Exiting.")
            return
        except Exception as e:
            logging.error(f"Trade Executor failed: {e}. Restarting...\n", exc_info=True)
            time.sleep(10)

def run_position_manager(trade_signal_queue: multiprocessing.Queue, order_execution_queue: multiprocessing.Queue):
    """
    Target function to run the PositionManager class in a resilient loop.
    Consumes from trade_signal_queue, produces for order_execution_queue.
    """

    # --- GRACEFUL SHUTDOWN HANDLER ---
    import signal
    
    def handle_shutdown_signal(signum, frame):
        # Raise KeyboardInterrupt, as it's handled by the loop
        logging.info(f"Shutdown signal ({signum}) received. Initiating graceful exit...")
        raise KeyboardInterrupt
            
    signal.signal(signal.SIGTERM, handle_shutdown_signal)
    # --- END GRACEFUL SHUTDOWN HANDLER ---

    log_file_path = os.path.join(LOGS_DIR, "position_manager.log")
    try:
        sys.stdout = open(log_file_path, 'a', buffering=1)
        sys.stderr = sys.stdout
    except Exception as e:
        print(f"Failed to open log file for PositionManager: {e}")

    setup_logging('normal', f"PositionManager")
    logging.info("\n--- Starting Position Manager process ---")

    while True:
        try:
            from position_manager import PositionManager
            
            manager = PositionManager(
                log_level="normal", 
                trade_signal_queue=trade_signal_queue,
                order_execution_queue=order_execution_queue
            )
            
            # --- REVERTED: Call manager.run() directly ---
            manager.run()

        except KeyboardInterrupt:
            logging.info("Position Manager interrupted. Exiting.")
            return
        except Exception as e:
            logging.error(f"Position Manager failed: {e}. Restarting...\n", exc_info=True)
            time.sleep(10)


def run_strategy(strategy_name: str, config: dict, trade_signal_queue: multiprocessing.Queue):
    """
    This function BECOMES the strategy runner. It is executed as a separate
    process and pushes signals to the shared queue.
    """
    # These imports only happen in the new, lightweight process
    import importlib
    import os
    import sys
    import time
    import logging
    import signal # <-- ADDED
    from logging_utils import setup_logging
    from strategies.base_strategy import BaseStrategy

    # --- GRACEFUL SHUTDOWN HANDLER ---
    def handle_shutdown_signal(signum, frame):
        # Raise KeyboardInterrupt, as it's handled by the loop
        try:
            logging.info(f"Shutdown signal ({signum}) received. Initiating graceful exit...")
        except NameError:
            print(f"[Strategy-{strategy_name}] Shutdown signal ({signum}) received. Initiating graceful exit...")
        raise KeyboardInterrupt
            
    signal.signal(signal.SIGTERM, handle_shutdown_signal)
    # --- END GRACEFUL SHUTDOWN HANDLER ---

    # --- Setup logging to file for this specific process ---
    log_file_path = os.path.join(LOGS_DIR, f"strategy_{strategy_name}.log")
    try:
        sys.stdout = open(log_file_path, 'a', buffering=1) # 1 = line buffering
        sys.stderr = sys.stdout
    except Exception as e:
        print(f"Failed to open log file for {strategy_name}: {e}")

    setup_logging('normal', f"Strategy-{strategy_name}")

    while True:
        try:
            logging.info(f"--- Starting strategy '{strategy_name}' ---")
            
            if 'class' not in config:
                logging.error(f"Strategy config for '{strategy_name}' is missing the 'class' key. Exiting.")
                return
            
            module_path, class_name = config['class'].rsplit('.', 1)
            module = importlib.import_module(module_path)
            StrategyClass = getattr(module, class_name)
            
            strategy = StrategyClass(strategy_name, config['parameters'], trade_signal_queue)

            if config.get("is_event_driven", False):
                logging.info(f"Starting EVENT-DRIVEN logic loop...")
                strategy.run_event_loop() # This is a blocking call
            else:
                logging.info(f"Starting POLLING logic loop...")
                strategy.run_polling_loop() # This is the original blocking call

        # --- REVERTED: Added back simple KeyboardInterrupt handler ---
        except KeyboardInterrupt:
            logging.info(f"Strategy {strategy_name} process stopping.")
            return
        except Exception as e:
            # --- REVERTED: Removed specific check for KeyboardInterrupt ---
            logging.error(f"Strategy '{strategy_name}' failed: {e}", exc_info=True)
            logging.info("Restarting strategy in 10 seconds...")
            time.sleep(10)


def run_dashboard_data_fetcher():
    """Target function to run the dashboard_data_fetcher.py script."""
    
    # --- GRACEFUL SHUTDOWN HANDLER ---
    import signal
    
    def handle_shutdown_signal(signum, frame):
        # Raise KeyboardInterrupt, as it's handled by the loop
        try:
            logging.info(f"Shutdown signal ({signum}) received. Initiating graceful exit...")
        except NameError:
            print(f"[DashboardDataFetcher] Shutdown signal ({signum}) received. Initiating graceful exit...")
        raise KeyboardInterrupt
            
    signal.signal(signal.SIGTERM, handle_shutdown_signal)
    # --- END GRACEFUL SHUTDOWN HANDLER ---
    
    log_file = os.path.join(LOGS_DIR, "dashboard_data_fetcher.log")
    while True:
        try:
            with open(log_file, 'a') as f:
                f.write(f"\n--- Starting Dashboard Data Fetcher at {datetime.now()} ---\n")
                subprocess.run([sys.executable, DASHBOARD_DATA_FETCHER_SCRIPT, "--log-level", "normal"], check=True, stdout=f, stderr=subprocess.STDOUT)
        except KeyboardInterrupt: # --- MODIFIED: Added to catch interrupt ---
            logging.info("Dashboard Data Fetcher stopping.")
            break
        except (subprocess.CalledProcessError, Exception) as e:
            with open(log_file, 'a') as f:
                f.write(f"\n--- PROCESS ERROR at {datetime.now()} ---\n")
                f.write(f"Dashboard Data Fetcher failed: {e}. Restarting...\n")
            time.sleep(10)


def run_indicators_fetcher():
    """Target function to run the indicators_fetcher.py script."""

    # --- GRACEFUL SHUTDOWN HANDLER ---
    import signal

    def handle_shutdown_signal(signum, frame):
        try:
            logging.info(f"Shutdown signal ({signum}) received. Initiating graceful exit...")
        except NameError:
            print(f"[IndicatorsFetcher] Shutdown signal ({signum}) received. Initiating graceful exit...")
        raise KeyboardInterrupt

    signal.signal(signal.SIGTERM, handle_shutdown_signal)
    # --- END GRACEFUL SHUTDOWN HANDLER ---

    log_file = os.path.join(LOGS_DIR, "indicators_fetcher.log")
    while True:
        try:
            with open(log_file, 'a') as f:
                f.write(f"\n--- Starting Indicators Fetcher at {datetime.now()} ---\n")
                subprocess.run([sys.executable, INDICATORS_FETCHER_SCRIPT, "--log-level", "normal"], check=True, stdout=f, stderr=subprocess.STDOUT)
        except KeyboardInterrupt:
            logging.info("Indicators Fetcher stopping.")
            break
        except (subprocess.CalledProcessError, Exception) as e:
            with open(log_file, 'a') as f:
                f.write(f"\n--- PROCESS ERROR at {datetime.now()} ---\n")
                f.write(f"Indicators Fetcher failed: {e}. Restarting...\n")
            time.sleep(10)


class MainApp:
    def __init__(self, coins_to_watch: list, processes: dict, strategy_configs: dict, shared_prices: dict):
        self.watched_coins = coins_to_watch
        self.shared_prices = shared_prices
        self.prices = {}
        self.background_processes = processes
        self.process_status = {}
        self.strategy_configs = strategy_configs
        self.strategy_statuses = {}
        self.indicators_status = {}
        self.account_data = None
        self.wallet_address = os.environ.get("MAIN_WALLET_ADDRESS")
        if self.wallet_address:
            self.info_client = Info(constants.MAINNET_API_URL, skip_ws=True)
        else:
            self.info_client = None
        self.renderer = DashboardRenderer(table_visibility={
            "market": True,
            "strategies": False,
            "indicators": True,
            "balances": True,
        })

    def read_prices(self):
        """Reads the latest prices directly from the shared memory dictionary."""
        try:
            # --- FIX: Use .copy() for thread-safe iteration ---
            self.prices = self.shared_prices.copy()
        except Exception as e:
            logging.debug(f"Could not read from shared prices dict: {e}")

    # --- REMOVED: read_market_caps method ---

    def read_strategy_statuses(self):
        """Reads the status JSON file for each enabled strategy."""
        enabled_statuses = {}
        for name, config in self.strategy_configs.items():
            if config.get("enabled", False):
                status_file = os.path.join("_data", f"strategy_status_{name}.json")
                if os.path.exists(status_file):
                    try:
                        with open(status_file, 'r', encoding='utf-8') as f:
                            enabled_statuses[name] = json.load(f)
                    except (IOError, json.JSONDecodeError):
                        enabled_statuses[name] = {"error": "Could not read status file."}
                else:
                    enabled_statuses[name] = {"current_signal": "Initializing..."}
        self.strategy_statuses = enabled_statuses

    def read_indicators_status(self):
        """Reads the indicators status JSON file."""
        status_file = os.path.join(LOGS_DIR, "indicators_status.json")
        if os.path.exists(status_file):
            try:
                with open(status_file, 'r', encoding='utf-8') as f:
                    self.indicators_status = json.load(f)
            except (IOError, json.JSONDecodeError):
                self.indicators_status = {}
        else:
            self.indicators_status = {}

    def read_account_data(self):
        """Fetches account balances and positions from Hyperliquid API."""
        if not self.wallet_address or not self.info_client:
            self.account_data = None
            return
        try:
            perp_state = self.info_client.user_state(self.wallet_address)
            spot_state = self.info_client.spot_user_state(self.wallet_address)

            margin_summary = perp_state.get('marginSummary', {})
            account_value = float(margin_summary.get('accountValue', 0))
            margin_used = float(margin_summary.get('totalMarginUsed', 0))
            utilization = (margin_used / account_value) * 100 if account_value > 0 else 0

            spot_balances = spot_state.get('balances', [])
            positions = perp_state.get('assetPositions', [])

            self.account_data = {
                'account_value': account_value,
                'margin_used': margin_used,
                'utilization': utilization,
                'spot_balances': spot_balances,
                'positions': positions,
            }

        except Exception as e:
            logging.error(f"Could not fetch account data: {e}")
            self.account_data = None

    def check_process_status(self):
        """Checks if the background processes are still running."""
        for name, process in self.background_processes.items():
            self.process_status[name] = "Running" if process.is_alive() else "STOPPED"

    def toggle_table(self, table_name, enabled=None):
        """Toggle a dashboard table's visibility at runtime."""
        return self.renderer.toggle_table(table_name, enabled)

    def display_dashboard(self):
        """Build and return the rich dashboard layout."""
        return self.renderer.build_layout(
            self.watched_coins,
            self.prices,
            COIN_DISPLAY_NAMES,
            self.strategy_statuses,
            self.strategy_configs,
            self.indicators_status,
            self.account_data
        )

    def run(self):
        """Main loop to read data, display dashboard, and check processes."""
        with Live(self.display_dashboard(), refresh_per_second=2, console=self.renderer.console) as live:
            while True:
                self.read_prices()
                self.read_strategy_statuses()
                self.read_indicators_status()
                self.read_account_data()
                live.update(self.display_dashboard())
                time.sleep(0.5)

if __name__ == "__main__":
    setup_logging('normal', 'MainApp')

    if not os.path.exists(LOGS_DIR):
        os.makedirs(LOGS_DIR)
        
    processes = {}
    # --- REVERTED: Removed process groups ---
    
    try:
        with open(STRATEGY_CONFIG_FILE, 'r') as f:
            strategy_configs = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError) as e:
        logging.error(f"Could not load strategies from '{STRATEGY_CONFIG_FILE}': {e}")
        sys.exit(1)
        
    # --- FIX: Hardcoded timeframes ---
    required_timeframes = [
        "3m", "5m", "15m", "30m", "1h", "2h", "4h", "8h", 
        "12h", "1d", "3d", "1w", "1M", "148m", "37m"
    ]
    logging.info(f"Using fixed timeframes for resampler: {required_timeframes}")
    
    with multiprocessing.Manager() as manager:
        shared_prices = manager.dict()
        # --- FIX: Create TWO queues ---
        trade_signal_queue = manager.Queue()
        order_execution_queue = manager.Queue()

        # --- REVERTED: All processes are daemon=True and in one dict ---
        
        # --- FIX: Pass WATCHED_COINS to the start_live_feed process ---
        # --- MODIFICATION: Set log level back to 'off' ---
        processes["Live Market Feed"] = multiprocessing.Process(
            target=start_live_feed, 
            args=(shared_prices, WATCHED_COINS, 'off'), 
            daemon=True
        )
        processes["Live Candle Fetcher"] = multiprocessing.Process(target=run_live_candle_fetcher, daemon=True)
        processes["Resampler"] = multiprocessing.Process(target=resampler_scheduler, args=(list(required_timeframes),), daemon=True)
        # --- REMOVED: Market Cap Fetcher Process ---
        processes["Dashboard Data"] = multiprocessing.Process(target=run_dashboard_data_fetcher, daemon=True)
        processes["Indicators"] = multiprocessing.Process(target=run_indicators_fetcher, daemon=True)

        processes["Position Manager"] = multiprocessing.Process(
            target=run_position_manager, 
            args=(trade_signal_queue, order_execution_queue),
            daemon=True
        )
        processes["Trade Executor"] = multiprocessing.Process(
            target=run_trade_executor, 
            args=(order_execution_queue,),
            daemon=True
        )
        
        for name, config in strategy_configs.items():
            if config.get("enabled", False):
                if 'class' not in config:
                    logging.error(f"Strategy '{name}' is missing 'class' key. Skipping.")
                    continue
                proc = multiprocessing.Process(target=run_strategy, args=(name, config, trade_signal_queue), daemon=True)
                processes[f"Strategy: {name}"] = proc # Add to strategy group
        
        # --- REVERTED: Removed combined dict ---

        for name, proc in processes.items():
            logging.info(f"Starting process '{name}'...")
            proc.start()
        
        time.sleep(3)

        app = MainApp(coins_to_watch=WATCHED_COINS, processes=processes, strategy_configs=strategy_configs, shared_prices=shared_prices)
        try:
            app.run()
        except KeyboardInterrupt:
            # --- MODIFIED: Staged shutdown ---
            logging.info("Shutting down...")

            strategy_procs = {}
            other_procs = {}
            for name, proc in processes.items():
                if name.startswith("Strategy:"):
                    strategy_procs[name] = proc
                else:
                    other_procs[name] = proc

            # --- 1. Terminate strategy processes ---
            logging.info("Shutting down strategy processes first...")
            for name, proc in strategy_procs.items():
                if proc.is_alive(): 
                    logging.info(f"Terminating process: '{name}'...")
                    proc.terminate()
            
            # --- 2. Wait for 5 seconds ---
            logging.info("Waiting 5 seconds for strategies to close...")
            time.sleep(5)
            
            # --- 3. Terminate all other processes ---
            logging.info("Shutting down remaining core processes...")
            for name, proc in other_procs.items():
                 if proc.is_alive(): 
                    logging.info(f"Terminating process: '{name}'...")
                    proc.terminate()
            
            # --- 4. Join all processes (strategies and others) ---
            logging.info("Waiting for all processes to join...")
            for name, proc in processes.items(): # Iterate over the original dict to get all
                if proc.is_alive(): 
                    logging.info(f"Waiting for process '{name}' to join...")
                    proc.join(timeout=5) # Wait up to 5 seconds
                    if proc.is_alive():
                        # If it's still alive, force kill
                        logging.warning(f"Process '{name}' did not terminate, forcing kill.")
                        proc.kill()
            # --- END MODIFIED ---
            
        logging.info("Shutdown complete.")
        sys.exit(0)