Hunters / app.py
omniverse1's picture
Update app.py
9aea858 verified
=== 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"}]
)
```