Update utils.py

utils.py

@@ -9,7 +9,6 @@ from plotly.subplots import make_subplots
 import spaces
 
 def get_indonesian_stocks():
-    """Get list of major Indonesian stocks"""
     return {
         "BBCA.JK": "Bank Central Asia",
         "BBRI.JK": "Bank BRI",
@@ -34,10 +33,7 @@ def get_indonesian_stocks():
     }
 
 def calculate_technical_indicators(data):
-    """Calculate various technical indicators"""
     indicators = {}
-    
-    # RSI (Relative Strength Index)
     def calculate_rsi(prices, period=14):
         delta = prices.diff()
         gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
@@ -45,13 +41,10 @@ def calculate_technical_indicators(data):
         rs = gain / loss
         rsi = 100 - (100 / (1 + rs))
         return rsi
-    
     indicators['rsi'] = {
         'current': calculate_rsi(data['Close']).iloc[-1],
         'values': calculate_rsi(data['Close'])
     }
-    
-    # MACD
     def calculate_macd(prices, fast=12, slow=26, signal=9):
         exp1 = prices.ewm(span=fast).mean()
         exp2 = prices.ewm(span=slow).mean()
@@ -59,7 +52,6 @@ def calculate_technical_indicators(data):
         signal_line = macd.ewm(span=signal).mean()
         histogram = macd - signal_line
         return macd, signal_line, histogram
-    
     macd, signal_line, histogram = calculate_macd(data['Close'])
     indicators['macd'] = {
         'macd': macd.iloc[-1],
@@ -69,64 +61,45 @@ def calculate_technical_indicators(data):
         'macd_values': macd,
         'signal_values': signal_line
     }
-    
-    # Bollinger Bands
     def calculate_bollinger_bands(prices, period=20, std_dev=2):
         sma = prices.rolling(window=period).mean()
         std = prices.rolling(window=period).std()
         upper_band = sma + (std * std_dev)
         lower_band = sma - (std * std_dev)
         return upper_band, sma, lower_band
-    
     upper, middle, lower = calculate_bollinger_bands(data['Close'])
     current_price = data['Close'].iloc[-1]
     bb_position = (current_price - lower.iloc[-1]) / (upper.iloc[-1] - lower.iloc[-1])
-    
     indicators['bollinger'] = {
         'upper': upper.iloc[-1],
         'middle': middle.iloc[-1],
         'lower': lower.iloc[-1],
         'position': 'UPPER' if bb_position > 0.8 else 'LOWER' if bb_position < 0.2 else 'MIDDLE'
     }
-    
-    # Moving Averages
     sma_20_series = data['Close'].rolling(20).mean()
     sma_50_series = data['Close'].rolling(50).mean()
-    
     indicators['moving_averages'] = {
-        'sma_20': sma_20_series.iloc[-1], # Current value
-        'sma_50': sma_50_series.iloc[-1], # Current value
+        'sma_20': sma_20_series.iloc[-1],
+        'sma_50': sma_50_series.iloc[-1],
         'sma_200': data['Close'].rolling(200).mean().iloc[-1],
         'ema_12': data['Close'].ewm(span=12).mean().iloc[-1],
         'ema_26': data['Close'].ewm(span=26).mean().iloc[-1],
-        
-        # ADDED: Full historical series for plotting
         'sma_20_values': sma_20_series,
         'sma_50_values': sma_50_series
     }
-    
-    # Volume indicators
     indicators['volume'] = {
         'current': data['Volume'].iloc[-1],
         'avg_20': data['Volume'].rolling(20).mean().iloc[-1],
         'ratio': data['Volume'].iloc[-1] / data['Volume'].rolling(20).mean().iloc[-1]
     }
-    
     return indicators
 
 def generate_trading_signals(data, indicators):
-    """Generate trading signals based on technical indicators"""
     signals = {}
-    
     current_price = data['Close'].iloc[-1]
-    
-    # Initialize scores
     buy_signals = 0
     sell_signals = 0
-    
     signal_details = []
-    
-    # RSI Signal
     rsi = indicators['rsi']['current']
     if rsi < 30:
         buy_signals += 1
@@ -136,8 +109,6 @@ def generate_trading_signals(data, indicators):
         signal_details.append(f"❌ RSI ({rsi:.1f}) - Overbought - SELL signal")
     else:
         signal_details.append(f"⚪ RSI ({rsi:.1f}) - Neutral")
-    
-    # MACD Signal
     macd_hist = indicators['macd']['histogram']
     if macd_hist > 0:
         buy_signals += 1
@@ -145,8 +116,6 @@ def generate_trading_signals(data, indicators):
     else:
         sell_signals += 1
         signal_details.append(f"❌ MACD Histogram ({macd_hist:.4f}) - Negative - SELL signal")
-    
-    # Bollinger Bands Signal
     bb_position = indicators['bollinger']['position']
     if bb_position == 'LOWER':
         buy_signals += 1
@@ -156,11 +125,8 @@ def generate_trading_signals(data, indicators):
         signal_details.append(f"❌ Bollinger Bands - Near upper band - SELL signal")
     else:
         signal_details.append("⚪ Bollinger Bands - Middle position")
-    
-    # Moving Averages Signal
     sma_20 = indicators['moving_averages']['sma_20']
     sma_50 = indicators['moving_averages']['sma_50']
-    
     if current_price > sma_20 > sma_50:
         buy_signals += 1
         signal_details.append(f"✅ Price above MA(20,50) - Bullish - BUY signal")
@@ -169,8 +135,6 @@ def generate_trading_signals(data, indicators):
         signal_details.append(f"❌ Price below MA(20,50) - Bearish - SELL signal")
     else:
         signal_details.append("⚪ Moving Averages - Mixed signals")
-    
-    # Volume Signal
     volume_ratio = indicators['volume']['ratio']
     if volume_ratio > 1.5:
         buy_signals += 0.5
@@ -180,22 +144,16 @@ def generate_trading_signals(data, indicators):
         signal_details.append(f"❌ Low volume ({volume_ratio:.1f}x avg) - Weakens SELL signal")
     else:
         signal_details.append(f"⚪ Normal volume ({volume_ratio:.1f}x avg)")
-    
-    # Determine overall signal
     total_signals = buy_signals + sell_signals
     signal_strength = (buy_signals / max(total_signals, 1)) * 100
-    
     if buy_signals > sell_signals:
         overall_signal = "BUY"
     elif sell_signals > buy_signals:
         overall_signal = "SELL"
     else:
         overall_signal = "HOLD"
-    
-    # Calculate support and resistance
     recent_high = data['High'].tail(20).max()
     recent_low = data['Low'].tail(20).min()
-    
     signals = {
         'overall': overall_signal,
         'strength': signal_strength,
@@ -204,15 +162,12 @@ def generate_trading_signals(data, indicators):
         'resistance': recent_high,
         'stop_loss': recent_low * 0.95 if overall_signal == "BUY" else recent_high * 1.05
     }
-    
     return signals
 
 def get_fundamental_data(stock):
-    """Get fundamental data for the stock"""
     try:
         info = stock.info
         history = stock.history(period="1d")
-        
         fundamental_info = {
             'name': info.get('longName', 'N/A'),
             'current_price': history['Close'].iloc[-1] if not history.empty else 0,
@@ -232,7 +187,6 @@ Book Value: {info.get('bookValue', 'N/A')}
 Price to Book: {info.get('priceToBook', 'N/A')}
             """.strip()
         }
-        
         return fundamental_info
     except Exception as e:
         print(f"Error getting fundamental data: {e}")
@@ -247,7 +201,6 @@ Price to Book: {info.get('priceToBook', 'N/A')}
         }
 
 def format_large_number(num):
-    """Format large numbers to readable format"""
     if num >= 1e12:
         return f"{num/1e12:.2f}T"
     elif num >= 1e9:
@@ -261,86 +214,50 @@ def format_large_number(num):
 
 @spaces.GPU(duration=120)
 def predict_prices(data, model, tokenizer, prediction_days=30):
-    """Predict future prices using Chronos-Bolt model"""
     try:
-        # Prepare data for prediction
         prices = data['Close'].values
         context_length = min(len(prices), 512)
-        
-        # Tokenize the input
         input_sequence = prices[-context_length:]
-        
-        # --- CRITICAL FIX: Simulate Quantization ---
-        # 1. Normalize prices (0 to 1)
         price_min = np.min(input_sequence)
         price_max = np.max(input_sequence)
-        
         if price_max == price_min:
-             normalized_sequence = np.zeros_like(input_sequence)
+            normalized_sequence = np.zeros_like(input_sequence)
         else:
-             normalized_sequence = (input_sequence - price_min) / (price_max - price_min)
-
-        # 2. Scale to a token space (max vocab size 4096) and convert to Long
-        VOCAB_SIZE = 4096 
-        # Convert to Long/Int to satisfy model embedding layer
-        token_indices = (normalized_sequence * (VOCAB_SIZE - 1)).astype(np.long) 
-        
-        # Create prediction input
-        # Pass tokens to the model
+            normalized_sequence = (input_sequence - price_min) / (price_max - price_min)
+        VOCAB_SIZE = getattr(model.config, "vocab_size", 2)
+        if VOCAB_SIZE == 2:
+            token_indices = (normalized_sequence > 0.5).astype(np.int64)
+        else:
+            token_indices = (normalized_sequence * (VOCAB_SIZE - 1)).astype(np.int64)
         prediction_input = torch.tensor(token_indices).unsqueeze(0).to(model.device)
-        # --- END CRITICAL FIX ---
-        
-        # Generate predictions
         with torch.no_grad():
-            # Use max_new_tokens for generation length.
-            # do_sample is necessary for generating probabilistic time-series forecasts
-            forecast = model.generate(
-                prediction_input,
-                max_new_tokens=prediction_days, 
-                do_sample=True 
-            )
-        
-        # Handle complex Chronos output: [batch_size, num_samples, prediction_length]
+            forecast = model.generate(prediction_input, max_new_tokens=prediction_days, do_sample=True)
         output_tensor = forecast[0] if isinstance(forecast, (tuple, list)) else forecast
-        
-        # Average across the samples and convert to a simple 1D numpy array
-        # Note: The output is still in TOKEN SPACE. We must INVERSE-SCALE it back to PRICE SPACE.
         predictions_tokens = output_tensor.float().mean(dim=1).squeeze().cpu().numpy()
-        
-        # --- CRITICAL INVERSE-SCALE FIX ---
-        # Inverse normalize the predicted tokens back to the price range
-        predictions = (predictions_tokens / (VOCAB_SIZE - 1)) * (price_max - price_min) + price_min
-        
-        # Handle case where predictions is a single scalar (convert to array for safety)
+        if VOCAB_SIZE == 2:
+            predictions = predictions_tokens * (price_max - price_min) + price_min
+        else:
+            predictions = (predictions_tokens / (VOCAB_SIZE - 1)) * (price_max - price_min) + price_min
         if predictions.ndim == 0:
-             predictions = np.array([predictions.item()])
-        
-        # Use actual prediction length from the output tensor
+            predictions = np.array([predictions.item()])
         pred_len = len(predictions)
-        
-        # Calculate prediction statistics
         last_price = prices[-1]
         predicted_high = np.max(predictions)
         predicted_low = np.min(predictions)
         predicted_mean = np.mean(predictions)
         change_pct = ((predicted_mean - last_price) / last_price) * 100
-        
         return {
             'values': predictions,
-            'dates': pd.date_range(
-                start=data.index[-1] + timedelta(days=1),
-                periods=pred_len, 
-                freq='D'
-            ),
+            'dates': pd.date_range(start=data.index[-1] + timedelta(days=1), periods=pred_len, freq='D'),
             'high_30d': predicted_high,
             'low_30d': predicted_low,
             'mean_30d': predicted_mean,
             'change_pct': change_pct,
             'summary': f"""
-AI Model: Amazon Chronos-Bolt (Simulated Quantization)
+AI Model: Amazon Chronos-Bolt (Binary Quantization)
 Prediction Period: {pred_len} days
 Expected Change: {change_pct:.2f}%
-Confidence: Medium (based on historical patterns)
+Confidence: Medium
 Note: AI predictions are for reference only and not financial advice
             """.strip()
         }
@@ -357,198 +274,37 @@ Note: AI predictions are for reference only and not financial advice
         }
 
 def create_price_chart(data, indicators):
-    """Create price chart with technical indicators"""
-    fig = make_subplots(
-        rows=3, cols=1,
-        shared_xaxes=True,
-        vertical_spacing=0.05,
-        subplot_titles=('Price & Moving Averages', 'RSI', 'MACD'),
-        row_width=[0.2, 0.2, 0.7]
-    )
-    
-    # Price and Moving Averages
-    fig.add_trace(
-        go.Candlestick(
-            x=data.index,
-            open=data['Open'],
-            high=data['High'],
-            low=data['Low'],
-            close=data['Close'],
-            name='Price'
-        ),
-        row=1, col=1
-    )
-    
-    # Add moving averages
-    fig.add_trace(
-        go.Scatter(
-            x=data.index,
-            y=indicators['moving_averages']['sma_20_values'], 
-            name='SMA 20',
-            line=dict(color='orange', width=1)
-        ),
-        row=1, col=1
-    )
-    
-    fig.add_trace(
-        go.Scatter(
-            x=data.index,
-            y=indicators['moving_averages']['sma_50_values'], 
-            name='SMA 50',
-            line=dict(color='blue', width=1)
-        ),
-        row=1, col=1
-    )
-    
-    # RSI
-    fig.add_trace(
-        go.Scatter(
-            x=data.index,
-            y=indicators['rsi']['values'],
-            name='RSI',
-            line=dict(color='purple')
-        ),
-        row=2, col=1
-    )
-    
+    fig = make_subplots(rows=3, cols=1, shared_xaxes=True, vertical_spacing=0.05, subplot_titles=('Price & Moving Averages', 'RSI', 'MACD'), row_width=[0.2, 0.2, 0.7])
+    fig.add_trace(go.Candlestick(x=data.index, open=data['Open'], high=data['High'], low=data['Low'], close=data['Close'], name='Price'), row=1, col=1)
+    fig.add_trace(go.Scatter(x=data.index, y=indicators['moving_averages']['sma_20_values'], name='SMA 20', line=dict(color='orange', width=1)), row=1, col=1)
+    fig.add_trace(go.Scatter(x=data.index, y=indicators['moving_averages']['sma_50_values'], name='SMA 50', line=dict(color='blue', width=1)), row=1, col=1)
+    fig.add_trace(go.Scatter(x=data.index, y=indicators['rsi']['values'], name='RSI', line=dict(color='purple')), row=2, col=1)
     fig.add_hline(y=70, line_dash="dash", line_color="red", row=2, col=1)
     fig.add_hline(y=30, line_dash="dash", line_color="green", row=2, col=1)
-    
-    # MACD
-    fig.add_trace(
-        go.Scatter(
-            x=data.index,
-            y=indicators['macd']['macd_values'], 
-            name='MACD',
-            line=dict(color='blue')
-        ),
-        row=3, col=1
-    )
-    
-    fig.add_trace(
-        go.Scatter(
-            x=data.index,
-            y=indicators['macd']['signal_values'], 
-            name='Signal',
-            line=dict(color='red')
-        ),
-        row=3, col=1
-    )
-    
-    fig.update_layout(
-        title='Technical Analysis Dashboard',
-        height=900,
-        showlegend=True,
-        xaxis_rangeslider_visible=False
-    )
-    
+    fig.add_trace(go.Scatter(x=data.index, y=indicators['macd']['macd_values'], name='MACD', line=dict(color='blue')), row=3, col=1)
+    fig.add_trace(go.Scatter(x=data.index, y=indicators['macd']['signal_values'], name='Signal', line=dict(color='red')), row=3, col=1)
+    fig.update_layout(title='Technical Analysis Dashboard', height=900, showlegend=True, xaxis_rangeslider_visible=False)
     return fig
 
 def create_technical_chart(data, indicators):
-    """Create technical indicators dashboard"""
-    fig = make_subplots(
-        rows=2, cols=2,
-        subplot_titles=('Bollinger Bands', 'Volume', 'Price vs MA', 'RSI Analysis'),
-        specs=[[{"secondary_y": False}, {"secondary_y": False}],
-               [{"secondary_y": False}, {"secondary_y": False}]]
-    )
-    
-    # Bollinger Bands
-    fig.add_trace(
-        go.Scatter(x=data.index, y=data['Close'], name='Price', line=dict(color='black')),
-        row=1, col=1
-    )
-    
-    # Volume
-    fig.add_trace(
-        go.Bar(x=data.index, y=data['Volume'], name='Volume', marker_color='lightblue'),
-        row=1, col=2
-    )
-    
-    # Price vs Moving Averages
-    fig.add_trace(
-        go.Scatter(x=data.index, y=data['Close'], name='Price', line=dict(color='black')),
-        row=2, col=1
-    )
-    
-    fig.add_trace(
-        go.Scatter(
-            x=data.index,
-            y=indicators['moving_averages']['sma_20_values'], 
-            name='SMA 20',
-            line=dict(color='orange', dash='dash')
-        ),
-        row=2, col=1
-    )
-    
-    fig.update_layout(
-        title='Technical Indicators Overview',
-        height=600,
-        showlegend=False
-    )
-    
+    fig = make_subplots(rows=2, cols=2, subplot_titles=('Bollinger Bands', 'Volume', 'Price vs MA', 'RSI Analysis'), specs=[[{"secondary_y": False}, {"secondary_y": False}], [{"secondary_y": False}, {"secondary_y": False}]])
+    fig.add_trace(go.Scatter(x=data.index, y=data['Close'], name='Price', line=dict(color='black')), row=1, col=1)
+    fig.add_trace(go.Bar(x=data.index, y=data['Volume'], name='Volume', marker_color='lightblue'), row=1, col=2)
+    fig.add_trace(go.Scatter(x=data.index, y=data['Close'], name='Price', line=dict(color='black')), row=2, col=1)
+    fig.add_trace(go.Scatter(x=data.index, y=indicators['moving_averages']['sma_20_values'], name='SMA 20', line=dict(color='orange', dash='dash')), row=2, col=1)
+    fig.update_layout(title='Technical Indicators Overview', height=600, showlegend=False)
     return fig
 
 def create_prediction_chart(data, predictions):
-    """Create prediction visualization"""
-    # Use len() check which works for both list and numpy array
     if not len(predictions['values']):
         return go.Figure()
-    
     fig = go.Figure()
-    
-    # Historical prices
-    fig.add_trace(
-        go.Scatter(
-            x=data.index[-60:],
-            y=data['Close'].values[-60:],
-            name='Historical Price',
-            line=dict(color='blue', width=2)
-        )
-    )
-    
-    # Predictions
-    fig.add_trace(
-        go.Scatter(
-            x=predictions['dates'],
-            y=predictions['values'],
-            name='AI Prediction',
-            line=dict(color='red', width=2, dash='dash')
-        )
-    )
-    
-    # Confidence interval (simple)
+    fig.add_trace(go.Scatter(x=data.index[-60:], y=data['Close'].values[-60:], name='Historical Price', line=dict(color='blue', width=2)))
+    fig.add_trace(go.Scatter(x=predictions['dates'], y=predictions['values'], name='AI Prediction', line=dict(color='red', width=2, dash='dash')))
     pred_std = np.std(predictions['values'])
     upper_band = predictions['values'] + (pred_std * 1.96)
     lower_band = predictions['values'] - (pred_std * 1.96)
-    
-    fig.add_trace(
-        go.Scatter(
-            x=predictions['dates'],
-            y=upper_band,
-            name='Upper Band',
-            line=dict(color='lightcoral', width=1),
-            fill=None
-        )
-    )
-    
-    fig.add_trace(
-        go.Scatter(
-            x=predictions['dates'],
-            y=lower_band,
-            name='Lower Band',
-            line=dict(color='lightcoral', width=1),
-            fill='tonexty',
-            fillcolor='rgba(255,182,193,0.2)'
-        )
-    )
-    
-    fig.update_layout(
-        title=f'Price Prediction - Next {len(predictions["dates"])} Days',
-        xaxis_title='Date',
-        yaxis_title='Price (IDR)',
-        hovermode='x unified',
-        height=500
-    )
-    
-    return fig
+    fig.add_trace(go.Scatter(x=predictions['dates'], y=upper_band, name='Upper Band', line=dict(color='lightcoral', width=1), fill=None))
+    fig.add_trace(go.Scatter(x=predictions['dates'], y=lower_band, name='Lower Band', line=dict(color='lightcoral', width=1), fill='tonexty', fillcolor='rgba(255,182,193,0.2)'))
+    fig.update_layout(title=f'Price Prediction - Next {len(predictions["dates"])} Days', xaxis_title='Date', yaxis_title='Price (IDR)', hovermode='x unified', height=500)
+    return fig