update utils-1.5

utils.py

@@ -14,12 +14,205 @@ from dataclasses import dataclass
 # Import configurations
 from config import IDX_STOCKS, IDX_MARKET_CONFIG 
 
-# ... (MarketStatusManager dan fungsi utilitas lainnya SAMA) ...
+
+# --- Market Status Manager ---
+
+@dataclass
+class MarketStatus:
+    """Data class to hold market status information"""
+    is_open: bool
+    status_text: str
+    next_trading_day: str
+    last_updated: str
+    time_until_open: str
+    time_until_close: str
+    current_time_et: str
+    market_name: str
+    market_type: str
+    market_symbol: str
+
+# Global configuration for market status updates
+MARKET_STATUS_UPDATE_INTERVAL_MINUTES = 10  # Update every 10 minutes
+
+class MarketStatusManager:
+    """Manages market status with periodic updates for the IDX market"""
+    
+    def __init__(self):
+        self.update_interval = MARKET_STATUS_UPDATE_INTERVAL_MINUTES * 60  # Convert to seconds
+        self._statuses = {}
+        self._lock = threading.Lock()
+        self._stop_event = threading.Event()
+        self._update_thread = None
+        self._start_update_thread()
+    
+    def _get_current_market_status(self, market_key: str = 'IDX_STOCKS') -> MarketStatus:
+        """Get current market status with detailed information for specific market"""
+        config = IDX_MARKET_CONFIG[market_key]
+        now = datetime.now(pytz.timezone('Asia/Jakarta')) # Use timezone-aware datetime for IDX
+        
+        market_tz = pytz.timezone(config['timezone'])
+        market_time = now.astimezone(market_tz)
+        
+        open_hour, open_minute = map(int, config['open_time'].split(':'))
+        close_hour, close_minute = map(int, config['close_time'].split(':'))
+        
+        market_open_time = market_time.replace(hour=open_hour, minute=open_minute, second=0, microsecond=0)
+        market_close_time = market_time.replace(hour=close_hour, minute=close_minute, second=0, microsecond=0)
+        
+        is_trading_day = market_time.weekday() in config['days']
+        
+        if not is_trading_day:
+            is_open = False
+            status_text = f"{config['name']} is closed (Weekend)"
+        elif market_open_time <= market_time <= market_close_time:
+            is_open = True
+            status_text = f"{config['name']} is currently open"
+        else:
+            is_open = False
+            if market_time < market_open_time:
+                status_text = f"{config['name']} is closed (Before opening)"
+            else:
+                status_text = f"{config['name']} is closed (After closing)"
+        
+        next_trading_day = self._get_next_trading_day(market_time, config['days'])
+        time_until_open = self._get_time_until_open(market_time, market_open_time, config)
+        time_until_close = self._get_time_until_close(market_time, market_close_time, config)
+        
+        return MarketStatus(
+            is_open=is_open,
+            status_text=status_text,
+            next_trading_day=next_trading_day.strftime('%Y-%m-%d'),
+            last_updated=market_time.strftime('%Y-%m-%d %H:%M:%S %Z'),
+            time_until_open=time_until_open,
+            time_until_close=time_until_close,
+            current_time_et=market_time.strftime('%H:%M:%S %Z'),
+            market_name=config['name'],
+            market_type=config['type'],
+            market_symbol=config['symbol']
+        )
+    
+    def _get_next_trading_day(self, current_time: datetime, trading_days: list) -> datetime:
+        """Get the next trading day for a specific market"""
+        next_day = current_time.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)
+        
+        while next_day.weekday() not in trading_days:
+            next_day += timedelta(days=1)
+        
+        return next_day
+    
+    def _get_time_until_open(self, current_time: datetime, market_open_time: datetime, config: dict) -> str:
+        """Calculate time until market opens"""
+        if current_time.weekday() not in config['days']:
+            days_until_next = 1
+            while (current_time + timedelta(days=days_until_next)).weekday() not in config['days']:
+                days_until_next += 1
+            
+            next_trading_day = current_time.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=days_until_next)
+            next_open = next_trading_day.replace(
+                hour=int(config['open_time'].split(':')[0]),
+                minute=int(config['open_time'].split(':')[1]),
+                second=0, microsecond=0
+            )
+            time_diff = next_open - current_time
+        else:
+            if current_time < market_open_time:
+                time_diff = market_open_time - current_time
+            else:
+                days_until_next = 1
+                while (current_time + timedelta(days=days_until_next)).weekday() not in config['days']:
+                    days_until_next += 1
+                
+                next_trading_day = current_time.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=days_until_next)
+                next_open = next_trading_day.replace(
+                    hour=int(config['open_time'].split(':')[0]),
+                    minute=int(config['open_time'].split(':')[1]),
+                    second=0, microsecond=0
+                )
+                time_diff = next_open - current_time
+        
+        return self._format_time_delta(time_diff)
+    
+    def _get_time_until_close(self, current_time: datetime, market_close_time: datetime, config: dict) -> str:
+        """Calculate time until market closes"""
+        if current_time.weekday() not in config['days']:
+            return "N/A (Weekend)"
+        
+        if current_time < market_close_time:
+            time_diff = market_close_time - current_time
+            return self._format_time_delta(time_diff)
+        else:
+            return "Market closed for today"
+    
+    def _format_time_delta(self, time_diff: timedelta) -> str:
+        """Format timedelta into human-readable string"""
+        total_seconds = int(time_diff.total_seconds())
+        if total_seconds < 0:
+            return "N/A"
+        
+        days = total_seconds // 86400
+        hours = (total_seconds % 86400) // 3600
+        minutes = (total_seconds % 3600) // 60
+        
+        if days > 0:
+            return f"{days}d {hours}h {minutes}m"
+        elif hours > 0:
+            return f"{hours}h {minutes}m"
+        else:
+            return f"{minutes}m"
+    
+    def _update_loop(self):
+        """Background thread loop for updating market status"""
+        while not self._stop_event.is_set():
+            try:
+                new_statuses = {}
+                market_key = 'IDX_STOCKS'
+                new_statuses[market_key] = self._get_current_market_status(market_key)
+                
+                with self._lock:
+                    self._statuses = new_statuses
+                time.sleep(self.update_interval)
+            except Exception as e:
+                print(f"Error updating market status: {str(e)}")
+                time.sleep(60)
+    
+    def _start_update_thread(self):
+        """Start the background update thread"""
+        if self._update_thread is None or not self._update_thread.is_alive():
+            self._stop_event.clear()
+            self._update_thread = threading.Thread(target=self._update_loop, daemon=True)
+            self._update_thread.start()
+    
+    def get_status(self, market_key: str = 'IDX_STOCKS') -> MarketStatus:
+        """Get current market status (thread-safe)"""
+        with self._lock:
+            if market_key not in self._statuses:
+                self._statuses[market_key] = self._get_current_market_status(market_key)
+            return self._statuses[market_key]
+    
+    def stop(self):
+        """Stop the update thread"""
+        self._stop_event.set()
+        if self._update_thread and self._update_thread.is_alive():
+            self._update_thread.join(timeout=5)
+
+# --- LAZY INITIALIZATION FIX ---
+_market_status_manager_instance = None
+
+def get_market_manager():
+    """Returns the singleton MarketStatusManager instance, initializing it if necessary."""
+    global _market_status_manager_instance
+    if _market_status_manager_instance is None:
+        _market_status_manager_instance = MarketStatusManager()
+    return _market_status_manager_instance
+
+# Hapus baris lama: market_status_manager = MarketStatusManager()
+# --- LAZY INITIALIZATION FIX ---
+
 
 # --- Utility Functions (Modified) ---
 
 def retry_yfinance_request(func, max_retries=3, initial_delay=1):
-    # ... (Fungsi ini SAMA) ...
+    # ... (SAMA) ...
     for attempt in range(max_retries):
         try:
             result = func()
@@ -39,7 +232,7 @@ def retry_yfinance_request(func, max_retries=3, initial_delay=1):
     return None
 
 def calculate_rsi(prices: pd.Series, period: int = 14) -> pd.Series:
-    # ... (Fungsi ini SAMA) ...
+    # ... (SAMA) ...
     prices = prices.ffill().bfill()
     delta = prices.diff()
     gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
@@ -48,7 +241,7 @@ def calculate_rsi(prices: pd.Series, period: int = 14) -> pd.Series:
     return 100 - (100 / (1 + rs))
 
 def calculate_macd(prices: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> Tuple[pd.Series, pd.Series]:
-    # ... (Fungsi ini SAMA) ...
+    # ... (SAMA) ...
     prices = prices.ffill().bfill()
     exp1 = prices.ewm(span=fast, adjust=False).mean()
     exp2 = prices.ewm(span=slow, adjust=False).mean()
@@ -57,7 +250,7 @@ def calculate_macd(prices: pd.Series, fast: int = 12, slow: int = 26, signal: in
     return macd, signal_line
 
 def calculate_bollinger_bands(prices: pd.Series, period: int = 20, std_dev: int = 2) -> Tuple[pd.Series, pd.Series, pd.Series]:
-    # ... (Fungsi ini SAMA) ...
+    # ... (SAMA) ...
     prices = prices.ffill().bfill()
     middle_band = prices.rolling(window=period).mean()
     std = prices.rolling(window=period).std()
@@ -66,7 +259,7 @@ def calculate_bollinger_bands(prices: pd.Series, period: int = 20, std_dev: int
     return upper_band, middle_band, lower_band
 
 def calculate_technical_indicators(df: pd.DataFrame) -> pd.DataFrame:
-    # ... (Fungsi ini SAMA) ...
+    # ... (SAMA) ...
     try:
         # Calculate moving averages
         df['SMA_20'] = df['Close'].rolling(window=20, min_periods=1).mean()
@@ -97,11 +290,11 @@ def calculate_technical_indicators(df: pd.DataFrame) -> pd.DataFrame:
 
 
 # --- Existing Utility Functions (Modified) ---
+
 # Hapus fungsi get_idx_stocks
 
 def fetch_stock_data(symbol, period_days):
-    """Fetch stock data from Yahoo Finance with retry and calculate indicators"""
-    # ... (Fungsi ini SAMA) ...
+    # ... (SAMA) ...
     try:
         end_date = datetime.now()
         start_date = end_date - timedelta(days=period_days + 30)
@@ -134,7 +327,7 @@ def fetch_stock_data(symbol, period_days):
         return None
 
 def prepare_timesfm_data(stock_data, use_volume=False):
-    # ... (Fungsi ini SAMA) ...
+    # ... (SAMA) ...
     data = stock_data['Close'].copy()
     
     data = data.fillna(method='ffill').fillna(method='bfill')
@@ -147,7 +340,7 @@ def prepare_timesfm_data(stock_data, use_volume=False):
 
 
 def create_forecast_plot(historical_data, forecast_dates, forecast_prices, symbol):
-    # ... (Fungsi ini SAMA) ...
+    # ... (SAMA) ...
     fig = make_subplots(
         rows=2, cols=1,
         shared_xaxes=True,
@@ -241,7 +434,7 @@ def create_forecast_plot(historical_data, forecast_dates, forecast_prices, symbo
     return fig
 
 def get_stock_info(symbol):
-    """Get additional stock information with retry"""
+    # ... (SAMA) ...
     try:
         ticker = yf.Ticker(symbol)
         info = retry_yfinance_request(lambda: ticker.info)