import warnings
# Suppress deprecation and future warnings from third-party libraries (like MLflow)
warnings.filterwarnings("ignore", category=FutureWarning)
import streamlit as st
import pandas as pd
import numpy as np
import os
import logging
import requests
import subprocess
import sys
from datetime import datetime
import altair as alt
import mlflow
# Configure logging
logging.basicConfig(level=logging.INFO)
# Determine robust file paths relative to script location
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_ROOT = os.path.abspath(os.path.join(SCRIPT_DIR, "..", ".."))
if PROJECT_ROOT not in sys.path:
sys.path.append(PROJECT_ROOT)
from src.models.export_forecasts import get_economic_factors
CLEAN_DATA_PATH = os.path.join(PROJECT_ROOT, "data", "processed", "clean_prices.csv")
DRIFT_STATUS_PATH = os.path.join(PROJECT_ROOT, "data", "processed", "drift_status.json")
MLFLOW_DIR = os.path.join(PROJECT_ROOT, "mlruns")
# Configure Streamlit page layout and title
st.set_page_config(
page_title="Sri Lankan Commodity Price Forecast",
page_icon="📈",
layout="wide",
initial_sidebar_state="expanded"
)
# Inject modern premium styling (Google Font Inter/Outfit and clean card UI)
st.markdown("""
""", unsafe_allow_html=True)
# Helper function to render a custom styled metric card
def render_metric_card(label, value_str, pct_change):
if pct_change > 0:
delta_class = "delta-up"
delta_str = f"▲ +{pct_change:.2f}%"
elif pct_change < 0:
delta_class = "delta-down"
delta_str = f"▼ {pct_change:.2f}%"
else:
delta_class = "delta-neutral"
delta_str = "● 0.00%"
return f"""
{label}
{value_str}
{delta_str} vs. Day 0
"""
# ----------------- DATA LOADING -----------------
@st.cache_data(ttl=60)
def load_historical_prices():
if os.path.exists(CLEAN_DATA_PATH):
df = pd.read_csv(CLEAN_DATA_PATH)
df["Date"] = pd.to_datetime(df["Date"])
return df
else:
st.warning(f"Cleaned CSV file not found at {CLEAN_DATA_PATH}. Using mock data.")
# Return empty df with columns to prevent crashes
return pd.DataFrame(columns=["Date", "Commodity", "Price"])
df_historical = load_historical_prices()
# Get list of commodities
if not df_historical.empty:
commodities_list = sorted(list(df_historical["Commodity"].unique()))
else:
commodities_list = ["Samba", "Kekulu", "Big Onion", "Potato", "Dried Chilli", "Coconut"]
# ----------------- SIDEBAR -----------------
st.sidebar.title("📈 CBSL Forecast Engine")
st.sidebar.markdown("---")
st.sidebar.subheader("🎯 Parameter Selection")
commodity_choice = st.sidebar.selectbox(
"Select Target Commodity",
commodities_list
)
st.sidebar.markdown("---")
st.sidebar.subheader("📡 Backend API Status")
# Check FastAPI Health Status
api_url = os.getenv("API_URL", "http://localhost:8000")
api_live = False
try:
health_resp = requests.get(f"{api_url}/", timeout=1)
if health_resp.status_code == 200 and health_resp.json().get("status") == "healthy":
api_live = True
except Exception:
pass
if api_live:
st.sidebar.markdown(
'Status: ● FastAPI Live',
unsafe_allow_html=True
)
else:
st.sidebar.markdown(
'Status: ● API Offline',
unsafe_allow_html=True
)
st.sidebar.warning(" FastAPI service is offline! Start it via terminal: `uvicorn src.models.api:app --reload` to fetch predictions.")
st.sidebar.markdown("---")
st.sidebar.subheader("🛠️ Pipeline Controls")
# Button to trigger drift check simulation and retraining
if st.sidebar.button("⚡ Run Retrain Simulation"):
st.sidebar.info("Running drift monitor with simulated inflation shock...")
# Run python src/models/monitor.py --simulate-shock
process = subprocess.Popen(
[sys.executable, "src/models/monitor.py", "--simulate-shock"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
cwd=PROJECT_ROOT
)
# Display running logs in the sidebar
log_placeholder = st.sidebar.empty()
log_text = ""
if process.stdout is not None:
while True:
output = process.stdout.readline()
if output == '' and process.poll() is not None:
break
if output:
log_text += output
# Display logs in small code block
log_placeholder.code(log_text[-1000:], language="bash")
rc = process.poll()
if rc == 0:
st.sidebar.success("🎉 Simulation completed!")
st.cache_data.clear() # Clear streamlit caches to load fresh runs/data
st.rerun()
else:
st.sidebar.error(f"❌ Execution failed (Code: {rc})")
st.sidebar.markdown("---")
st.sidebar.info("Built with streamlit to support continuous integration drift metrics.")
# ----------------- MAIN PANEL -----------------
st.markdown('📈 Sri Lankan Commodity Price Forecasting
', unsafe_allow_html=True)
st.markdown('MLOps Automated Daily Scraper & XGBoost Recursive Forecast serving Hugging Face Spaces
', unsafe_allow_html=True)
# ----------------- TABS CREATION -----------------
tab_forecast, tab_history, tab_mlops = st.tabs([
"🔮 Forecast & Economic Indicators",
"📊 Historical Trends & Analytics",
"🛡️ MLOps Control Room & Model Registry"
])
# ----------------- FETCH PREDICTIONS -----------------
forecast_prices = []
recent_lag_prices = []
forecast_error_msg = None
# Filter historical data for target commodity
df_comm = df_historical[df_historical["Commodity"].str.lower() == commodity_choice.lower()].sort_values("Date")
if not df_comm.empty:
recent_prices = df_comm.tail(7)
if len(recent_prices) == 7:
recent_lag_prices = recent_prices["Price"].tolist()
last_actual_price = recent_lag_prices[-1]
last_actual_date = recent_prices["Date"].max()
else:
# Fallback mocks if dataset has too few points
recent_lag_prices = [220.0] * 7
last_actual_price = 220.0
last_actual_date = datetime.now()
forecast_error_msg = f"Insufficient history (need 7 days, found {len(recent_prices)}) to build prediction lag."
else:
recent_lag_prices = [220.0] * 7
last_actual_price = 220.0
last_actual_date = datetime.now()
forecast_error_msg = "No historical price series found for this commodity."
# Call FastAPI predicted forecast
if api_live and not forecast_error_msg:
try:
pred_resp = requests.post(
f"{api_url}/predict",
json={
"commodity": commodity_choice,
"lag_prices": recent_lag_prices
},
timeout=3
)
if pred_resp.status_code == 200:
forecast_prices = pred_resp.json().get("7_day_forecast", [])
else:
forecast_error_msg = f"FastAPI returned error {pred_resp.status_code}: {pred_resp.json().get('detail')}"
except Exception as e:
forecast_error_msg = f"Connection timeout fetching prediction: {e}"
else:
if not forecast_error_msg:
forecast_error_msg = "FastAPI backend server is currently offline."
# =========================================================================
# TAB 1: FORECAST & ECONOMIC INDICATORS
# =========================================================================
with tab_forecast:
if forecast_error_msg:
st.warning(f"⚠️ Forecast unavailable: {forecast_error_msg}")
if forecast_prices:
# 1. Metric Cards Row
pct_tomorrow = ((forecast_prices[0] - last_actual_price) / last_actual_price) * 100
pct_day7 = ((forecast_prices[-1] - last_actual_price) / last_actual_price) * 100
col1, col2, col3 = st.columns(3)
with col1:
st.markdown(
render_metric_card("Current Price (Day 0)", f"LKR {last_actual_price:.2f}", 0.0),
unsafe_allow_html=True
)
with col2:
st.markdown(
render_metric_card("Tomorrow's Prediction", f"LKR {forecast_prices[0]:.2f}", pct_tomorrow),
unsafe_allow_html=True
)
with col3:
st.markdown(
render_metric_card("Day 7 Forecast", f"LKR {forecast_prices[-1]:.2f}", pct_day7),
unsafe_allow_html=True
)
# 2. Stitch together continuous plot (Last 14 days Actuals + 7 Days Forecast)
df_act_plot = df_comm.tail(14).copy()
df_act_plot = df_act_plot[["Date", "Price"]].rename(columns={"Price": "Price (LKR)"})
df_act_plot["Type"] = "Actual"
# Build future dates
future_dates = [last_actual_date + pd.Timedelta(days=i) for i in range(1, 8)]
df_fore_plot = pd.DataFrame({
"Date": future_dates,
"Price (LKR)": forecast_prices,
"Type": "Forecast"
})
# Connect transition point (last actual becomes first point of forecast series to bridge line gap)
transition_row = pd.DataFrame({
"Date": [last_actual_date],
"Price (LKR)": [last_actual_price],
"Type": ["Forecast"]
})
df_fore_plot = pd.concat([transition_row, df_fore_plot], ignore_index=True)
df_combined = pd.concat([df_act_plot, df_fore_plot], ignore_index=True)
# Plot continuous line using Altair
st.subheader("🔮 Unified 7-Day Recursive Forecast Trend")
line_chart = alt.Chart(df_combined).mark_line(point=True).encode(
x=alt.X('Date:T', title='Date', axis=alt.Axis(format='%b %d', labelAngle=-30)),
y=alt.Y('Price (LKR):Q', title='Price (LKR/kg)', scale=alt.Scale(zero=False)),
color=alt.Color('Type:N', scale=alt.Scale(domain=['Actual', 'Forecast'], range=['#2563eb', '#f59e0b']), title='Status'),
strokeDash=alt.condition(
alt.datum.Type == 'Forecast',
alt.value([6, 4]), # dashed line
alt.value([0]) # solid line
),
tooltip=['Date:T', 'Price (LKR):Q', 'Type:N']
).properties(
height=400,
title=f"7-Day Continuous Pricing Trend for {commodity_choice}"
).interactive()
st.altair_chart(line_chart, use_container_width=True)
# 3. Macroeconomic Factors
st.markdown("---")
st.subheader("🌐 Correlated Economic Factors Outlook")
st.markdown("Predicted exchange rates and fuel prices computed dynamically for forecast days:")
ec_col1, ec_col2 = st.columns(2)
# Calculate factors for future dates
forecast_dates_str = [d.strftime("%Y-%m-%d") for d in future_dates]
eco_factors = [get_economic_factors(d) for d in future_dates]
exchange_rates = [e[0] for e in eco_factors]
fuel_prices = [e[1] for e in eco_factors]
df_eco = pd.DataFrame({
"Date": forecast_dates_str,
"USD/LKR Rate": exchange_rates,
"Petrol 92 Price (LKR/L)": fuel_prices
})
with ec_col1:
st.markdown("##### 💵 USD/LKR Exchange Rate Target")
eco_chart1 = alt.Chart(df_eco).mark_line(color='#10b981', point=True).encode(
x=alt.X('Date:T', title='Forecast Date'),
y=alt.Y('USD/LKR Rate:Q', title='Rate (LKR)', scale=alt.Scale(zero=False)),
tooltip=['Date:T', 'USD/LKR Rate:Q']
).properties(height=250)
st.altair_chart(eco_chart1, use_container_width=True)
with ec_col2:
st.markdown("##### ⛽ CPC Petrol 92 Retail Price")
eco_chart2 = alt.Chart(df_eco).mark_line(color='#ef4444', point=True).encode(
x=alt.X('Date:T', title='Forecast Date'),
y=alt.Y('Petrol 92 Price (LKR/L):Q', title='Price (LKR)', scale=alt.Scale(zero=False)),
tooltip=['Date:T', 'Petrol 92 Price (LKR/L):Q']
).properties(height=250)
st.altair_chart(eco_chart2, use_container_width=True)
else:
st.info("💡 Start the FastAPI engine using terminal commands to render recursive predictions.")
# =========================================================================
# TAB 2: HISTORICAL TRENDS & ANALYTICS
# =========================================================================
with tab_history:
if df_historical.empty:
st.warning("No historical dataset loaded.")
else:
# User controls for dates
min_date = df_comm["Date"].min()
max_date = df_comm["Date"].max()
h_col1, h_col2 = st.columns([3, 1])
with h_col1:
st.subheader("📅 Filter Historical Timeline")
selected_dates = st.date_input(
"Select Date Range",
value=(min_date, max_date),
min_value=min_date,
max_value=max_date
)
# Parse filter
if isinstance(selected_dates, tuple) and len(selected_dates) == 2:
start_d, end_d = selected_dates
df_filtered = df_comm[(df_comm["Date"] >= pd.to_datetime(start_d)) & (df_comm["Date"] <= pd.to_datetime(end_d))].copy()
else:
df_filtered = df_comm.copy()
# Calculate moving averages
df_filtered["7-Day MA"] = df_filtered["Price"].rolling(window=7, min_periods=1).mean()
df_filtered["30-Day MA"] = df_filtered["Price"].rolling(window=30, min_periods=1).mean()
# Melt columns for cleaner multi-line charting
df_melted = df_filtered.melt(
id_vars=["Date"],
value_vars=["Price", "7-Day MA", "30-Day MA"],
var_name="Series",
value_name="LKR"
).dropna()
# Draw historical plot
hist_line = alt.Chart(df_melted).mark_line().encode(
x=alt.X('Date:T', title='Timeline'),
y=alt.Y('LKR:Q', title='Price per kg (LKR)', scale=alt.Scale(zero=False)),
color=alt.Color('Series:N', scale=alt.Scale(
domain=['Price', '7-Day MA', '30-Day MA'],
range=['#2563eb', '#10b981', '#ef4444']
), title='Legend'),
tooltip=['Date:T', 'Series:N', 'LKR:Q']
).properties(
height=400,
title=f"Historical Price Series with Technical Moving Averages for {commodity_choice}"
).interactive()
st.altair_chart(hist_line, use_container_width=True)
# Statistical summary cards
st.markdown("---")
st.subheader("📊 Summary Statistics")
if not df_filtered.empty:
prices = df_filtered["Price"]
stat_min = prices.min()
stat_max = prices.max()
stat_avg = prices.mean()
stat_std = prices.std()
s_col1, s_col2, s_col3, s_col4 = st.columns(4)
s_col1.metric("Min Price", f"LKR {stat_min:.2f}")
s_col2.metric("Max Price", f"LKR {stat_max:.2f}")
s_col3.metric("Average Price", f"LKR {stat_avg:.2f}")
s_col4.metric("Volatility (Std Dev)", f"{stat_std:.2f}" if not pd.isna(stat_std) else "0.00")
# Download CSV
csv_data = df_filtered.to_csv(index=False).encode('utf-8')
st.download_button(
label="📥 Download Cleaned Time-Series Data (CSV)",
data=csv_data,
file_name=f"{commodity_choice}_cleaned_prices.csv",
mime="text/csv"
)
# =========================================================================
# TAB 3: MLOPS CONTROL ROOM & MODEL REGISTRY
# =========================================================================
with tab_mlops:
st.subheader("🛡️ Real-Time Concept Drift Monitor")
# 1. Load drift status JSON
if os.path.exists(DRIFT_STATUS_PATH):
try:
import json
with open(DRIFT_STATUS_PATH, "r") as f:
drift_status = json.load(f)
dr_col1, dr_col2, dr_col3 = st.columns(3)
dr_col1.metric("Latest Drift Checked", drift_status.get("timestamp"))
# Support both statistical KS test format and legacy mean-based percentage drift
is_ks_test = "p_value" in drift_status
if is_ks_test:
ks_stat = drift_status.get("drift_score", 0.0)
p_val = drift_status.get("p_value", 1.0)
significance_level = drift_status.get("drift_threshold", 0.05)
dr_col2.metric("KS Statistic", f"{ks_stat:.4f}", help="Kolmogorov-Smirnov distance between incoming 7-day prices and historical baseline.")
dr_col3.metric("p-value (Significance)", f"{p_val:.4f}", help=f"Drift is detected if p-value < significance level ({significance_level}).")
if drift_status.get("drift_detected"):
st.error(f"🚨 **Alert: Concept Drift Detected (p-value: {p_val:.4f} < {significance_level:.2f})**! The Kolmogorov-Smirnov test confirms a statistically significant change in price distributions. Automated retraining was triggered.")
else:
st.success(f"✅ **System Stable**: No significant drift detected (p-value: {p_val:.4f} >= {significance_level:.2f}). Prices follow the baseline distribution.")
else:
drift_score_pct = drift_status.get("drift_score", 0) * 100
drift_thresh_pct = drift_status.get("drift_threshold", 0.15) * 100
dr_col2.metric("Current Drift Score", f"{drift_score_pct:.2f}%", help="Deviation of incoming 7-day average from historical mean.")
dr_col3.metric("Drift Alert Threshold", f"{drift_thresh_pct:.2f}%")
if drift_status.get("drift_detected"):
st.error(f"🚨 **Alert: Concept Drift Detected ({drift_score_pct:.2f}% > {drift_thresh_pct:.2f}%)**! The automated retraining pipeline was triggered to adapt to market shocks.")
else:
st.success(f"✅ **System Stable**: Current drift ({drift_score_pct:.2f}%) is within safe thresholds. XGBoost forecast model parameters are calibrated.")
with st.expander("🔍 View Technical Drift Details"):
st.json(drift_status)
except Exception as e:
st.warning(f"Error parsing drift status logs: {e}")
else:
st.warning("⚠️ No drift logs discovered. Click '⚡ Run Retrain Simulation' in the sidebar to perform a monitoring sweep.")
st.markdown("---")
st.subheader("📦 MLflow Local Model Registry Logs")
# 2. Load MLflow experiment metadata
mlflow.set_tracking_uri(f"file:{MLFLOW_DIR}")
try:
experiment = mlflow.get_experiment_by_name("SL_Commodity_Forecasting")
if experiment:
runs_raw = mlflow.search_runs(
experiment_ids=[experiment.experiment_id],
order_by=["start_time DESC"]
)
# Ensure type safety and runtime verification for the runs data
if isinstance(runs_raw, pd.DataFrame) and not runs_raw.empty:
# Filter runs programmatically since MLflow query parser does not support OR/IN operators
target_runs = ['XGBoost_Candidate', 'LinearRegression_Baseline', 'ARIMA_Baseline']
if "tags.mlflow.runName" in runs_raw.columns:
runs_raw = runs_raw[runs_raw["tags.mlflow.runName"].isin(target_runs)]
if not runs_raw.empty:
runs_df = runs_raw
st.write(f"Showing recent model training runs logged to experiment: **{experiment.name}**")
# Cleanup DataFrame for visualization
display_cols = []
rename_dict = {}
for c in runs_df.columns:
if c == "run_id":
display_cols.append(c)
rename_dict[c] = "Run ID"
elif "runName" in c:
display_cols.append(c)
rename_dict[c] = "Model/Run Name"
elif c == "start_time":
display_cols.append(c)
rename_dict[c] = "Timestamp"
elif "metrics.rmse" in c:
display_cols.append(c)
rename_dict[c] = "RMSE (Test)"
elif "metrics.mae" in c:
display_cols.append(c)
rename_dict[c] = "MAE (Test)"
elif "params.learning_rate" in c:
display_cols.append(c)
rename_dict[c] = "Learning Rate"
elif "params.max_depth" in c:
display_cols.append(c)
rename_dict[c] = "Max Depth"
df_runs_styled = runs_df[display_cols].rename(columns=rename_dict).copy()
# Format Timestamp column
if "Timestamp" in df_runs_styled.columns:
df_runs_styled["Timestamp"] = pd.to_datetime(df_runs_styled["Timestamp"]).dt.strftime("%Y-%m-%d %H:%M:%S")
st.dataframe(df_runs_styled, use_container_width=True)
else:
st.info("No runs logged in MLflow registry.")
else:
st.info("No runs logged in MLflow registry.")
else:
st.info("Experiment 'SL_Commodity_Forecasting' has not been created yet. Run model training.")
except Exception as e:
st.warning(f"Unable to read local MLflow tracking files: {e}")
# ----------------- FOOTER -----------------
st.markdown("---")
st.markdown(
''
'Case Study: Sri Lankan Commodity Price Forecasting • Built using Streamlit, MLflow, and FastAPI serving local models. '
'All historical data sourced from CBSL Daily Price Reports.
',
unsafe_allow_html=True
)