=== app.py === ```python import gradio as gr import yfinance as yf import pandas as pd import numpy as np import torch from chronos import ChronosPipeline import plotly.graph_objects as go from datetime import datetime, timedelta import warnings # Suppress warnings for cleaner output warnings.filterwarnings("ignore") # --- Global Model Loading --- print("Loading Chronos Model...") try: pipeline = ChronosPipeline.from_pretrained( "amazon/chronos-t5-small", device_map="auto", torch_dtype=torch.float32, # Use float32 for CPU compatibility ) print("Model loaded successfully on CPU.") except Exception as e: print(f"Error loading model: {e}") pipeline = None # --- Constants & Configuration --- IDX_WATCHLIST = ["BBCA.JK", "BBRI.JK", "GOTO.JK", "ANTM.JK", "TLKM.JK", "ASII.JK", "UNTR.JK", "ADRO.JK"] HISTORY_DAYS = 60 PREDICTION_HORIZON = 1 # --- Helper Functions --- def fetch_data(ticker, period="60d"): """ Fetch historical data from Yahoo Finance. Automatically appends .JK if missing for Indonesian stocks. """ ticker_clean = ticker.upper().strip() if not ticker_clean.endswith(".JK"): ticker_clean += ".JK" try: stock = yf.Ticker(ticker_clean) df = stock.history(period=period) if df.empty: return None, f"No data found for {ticker_clean}" return df, ticker_clean except Exception as e: return None, str(e) def predict_price(data, pipeline_model): """ Perform Chronos inference. Returns a dictionary with P10, P50, P90 predictions. """ if pipeline_model is None: return None try: # Chronos expects a 1D context array context = data["Close"].values[-HISTORY_DAYS:].tolist() # Predict prediction = pipeline_model.predict( context, prediction_length=PREDICTION_HORIZON ) # Extract quantiles: index 0 is P10, 1 is P50, 2 is P90 in Chronos output usually # Depending on version, we might need to check shape. # Chronos-T5 output shape is (num_samples, prediction_length) # We take the median as P50 and calculate actual percentiles from samples samples = prediction[0] # Get first sample dimension p10 = np.percentile(samples, 10, axis=0) p50 = np.percentile(samples, 50, axis=0) p90 = np.percentile(samples, 90, axis=0) return { "p10": p10[0], "p50": p50[0], "p90": p90[0] } except Exception as e: print(f"Prediction error: {e}") return None def calculate_metrics(last_close, preds, df_history): """ Calculate Gain %, Volume Surge, and Confidence. """ gain_pct = ((preds['p50'] - last_close) / last_close) * 100 # Volume Surge: Current Volume vs 20-day average current_vol = df_history['Volume'].iloc[-1] avg_vol = df_history['Volume'].tail(20).mean() vol_surge = ((current_vol - avg_vol) / avg_vol) * 100 if avg_vol > 0 else 0 # Confidence: Inverse of() spread (P90 - P10) relative to price spread = preds['p90'] - preds['p10'] confidence = 100 - ((spread / last_close) * 100) confidence = max(0, min(100, confidence)) # Clamp between 0 and 100 return gain_pct, vol_surge, confidence # --- Gradio Logic --- def scan_market(): """ Main logic for Tab 1: Screener. """ if pipeline is None: return pd.DataFrame([{"Error": "Model not loaded"}]) results = [] for ticker in IDX_WATCHLIST: try: df, ticker_clean = fetch_data(ticker) if df is None or len(df) < HISTORY_DAYS: continue last_close = df['Close'].iloc[-1] preds = predict_price(df, pipeline) if preds: gain, surge, conf = calculate_metrics(last_close, preds, df) results.append({ "Ticker": ticker_clean.replace(".JK", ""), "Last Close": round(last_close, 2), "Predicted High": round(preds['p90'], 2), # Using P90 as potential high "Gain %": round(gain, 2), "Confidence": round(conf, 1), "Volume Surge %": round(surge, 2) }) except Exception as e: print(f"Error processing {ticker}: {e}") continue if not results: return pd.DataFrame([{"Message": "No data processed successfully"}]) results_df = pd.DataFrame(results) # Sort by Gain % descending results_df = results_df.sort_values(by="Gain %", ascending=False) return results_df def analyze_stock(ticker_input): """ Main logic for Tab 2: Analyzer. """ if not ticker_input: return None df, ticker_clean = fetch_data(ticker_input, period="2y") if df is None: return None # Return None for plot, Gradio handles error display usually or we could return text if len(df) < HISTORY_DAYS: return None # Predict preds = predict_price(df, pipeline) if not preds: return None # Prepare Data for Plotting last_date = df.index[-1] next_date = last_date + timedelta(days=1) # Historical Trace hist_trace = go.Scatter( x=df.index, y=df['Close'], mode='lines', name='Historical Price', line=dict(color='gray', width=2) ) # Prediction Trace (P50) pred_trace = go.Scatter( x=[last_date, next_date], y=[df['Close'].iloc[-1], preds['p50']], mode='lines+markers', name='P50 Forecast', line=dict(color='green', width=2, dash='dash') ) # Uncertainty Cloud (P10 to P90) cloud_x = [last_date, next_date, next_date, last_date] cloud_y = [ df['Close'].iloc[-1], preds['p10'], preds['p90'], df['Close'].iloc[-1] ] cloud_trace = go.Scatter( x=cloud_x, y=cloud_y, mode='lines', fill='toself', fillcolor='rgba(0, 100, 80, 0.2)', # Light green transparent line=dict(color='rgba(0,0,0,0)'), name='P10-P90 Range' ) layout = go.Layout( title=f"Price Prediction: {ticker_clean}", xaxis_title="Date", yaxis_title="Price (IDR)", hovermode='x unified', template="plotly_white" ) fig = go.Figure(data=[hist_trace, cloud_trace, pred_trace], layout=layout) return fig # --- Gradio Interface Setup --- # Gradio 6: Blocks() takes NO parameters with gr.Blocks() as demo: gr.Markdown( """ # 🇮🇩 IDX Stock Screener (Chronos AI) Built with [anycoder](https://huggingface.co/spaces/akhaliq/anycoder) Predict Indonesian stock movements using Amazon's Chronos-T5 Time Series model. """ ) with gr.Tabs(): with gr.TabItem("Market Screener"): gr.Markdown("### Scan top liquid IDX stocks for potential gains.") with gr.Row(): scan_btn = gr.Button("Scan Market (Watchlist)", variant="primary", size="lg") screener_output = gr.Dataframe( label="Screener Results", datatype=["str", "number", "number", "number", "number", "number"], interactive=False ) # Gradio 6: Use api_visibility in event listeners scan_btn.click( fn=scan_market, inputs=[], outputs=screener_output, api_visibility="public" ) with gr.TabItem("Stock Analyzer"): gr.Markdown("### Detailed analysis and charting for specific stocks.") with gr.Row(): ticker_input = gr.Textbox( label="Stock Ticker (e.g., BBRI)", placeholder="Enter ticker code...", scale=3 ) analyze_btn = gr.Button("Analyze", variant="primary", scale=1) plot_output = gr.Plot(label="Price Forecast Chart") analyze_btn.click( fn=analyze_stock, inputs=[ticker_input], outputs=[plot_output], api_visibility="public" ) # Launch app if __name__ == "__main__": # Gradio 6: ALL app parameters (theme, footer_links) go in launch() demo.launch( theme=gr.themes.Soft( primary_hue="blue", secondary_hue="indigo", neutral_hue="slate", font=gr.themes.GoogleFont("Inter"), text_size="lg", spacing_size="lg", radius_size="md" ), footer_links=[{"label": "Built with anycoder", "url": "https://huggingface.co/spaces/akhaliq/anycoder"}] ) ```