Spaces:
Runtime error
Runtime error
| === 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"}] | |
| ) | |
| ``` |