Create app.py

app.py

@@ -0,0 +1,126 @@
+from fastapi import FastAPI
+from apscheduler.schedulers.background import BackgroundScheduler
+import pandas as pd
+import requests
+import xgboost as xgb
+from datetime import datetime, timedelta
+import os
+import matplotlib.pyplot as plt
+
+app = FastAPI()
+
+# Scheduler
+scheduler = BackgroundScheduler()
+scheduler.start()
+
+# Load the trained model
+model = xgb.XGBRegressor()
+model.load_model("electricity_price_model.json")
+
+# Constants
+WEATHER_API = "https://api.open-meteo.com/v1/forecast"
+ELECTRICITY_PRICE_API = "https://www.elprisetjustnu.se/api/v1/prices"
+ENERGY_CHARTS_API = "https://api.energy-charts.info/public_power?"
+PREDICTIONS_FILE = "predicted_prices.csv"
+
+# Fetch weather data for the next day
+def fetch_weather_data_for_tomorrow():
+    tomorrow = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d')
+    params = {
+        "latitude": 59.3293,
+        "longitude": 18.0686,
+        "daily": "temperature_2m_mean,precipitation_sum,wind_speed_10m_max,wind_direction_10m_dominant",
+        "start_date": tomorrow,
+        "end_date": tomorrow,
+        "timezone": "Europe/Stockholm"
+    }
+    response = requests.get(WEATHER_API, params=params)
+    response.raise_for_status()
+    data = response.json()["daily"]
+    return pd.DataFrame(data)
+
+# Fetch energy production data for the current day
+def fetch_energy_production_data():
+    today = datetime.now().strftime('%Y-%m-%d')
+    params = {"country": "se", "start": today, "end": today}
+    response = requests.get(ENERGY_CHARTS_API, params=params)
+    response.raise_for_status()
+    data = response.json()
+
+    if "production_types" in data:
+        production_data = {
+            "unix_seconds": data["unix_seconds"],
+            **{ptype["name"]: ptype["data"] for ptype in data["production_types"]}
+        }
+        energy_df = pd.DataFrame(production_data)
+        energy_df = energy_df.rename(columns={"unix_seconds": "time"})
+        energy_df["time"] = pd.to_datetime(energy_df["time"], unit="s", errors="coerce").dt.tz_localize(None)
+        return energy_df
+    else:
+        return pd.DataFrame()
+
+# Fetch electricity prices for the current day
+def fetch_current_electricity_prices():
+    today = datetime.now().strftime('%Y/%m-%d')
+    url = f"{ELECTRICITY_PRICE_API}/{today}_SE3.json"
+    response = requests.get(url)
+    response.raise_for_status()
+    data = response.json()
+
+    electricity_df = pd.DataFrame(data)
+    electricity_df = electricity_df.rename(columns={"time_start": "time"})
+    electricity_df["time"] = pd.to_datetime(electricity_df["time"], errors="coerce").dt.tz_localize(None)
+    return electricity_df
+
+# Prepare dataset for prediction
+def prepare_prediction_data():
+    energy_data = fetch_energy_production_data()
+    electricity_data = fetch_current_electricity_prices()
+    weather_data = fetch_weather_data_for_tomorrow()
+
+    dataset = pd.merge(energy_data, electricity_data, on="time", how="inner")
+    dataset = pd.merge(dataset, weather_data, on="time", how="outer")
+    dataset = dataset.dropna()
+    return dataset
+
+# Predict electricity prices for the next day
+def predict_next_day_price():
+    dataset = prepare_prediction_data()
+    X = dataset.drop(["SEK_per_kWh", "time"], axis=1, errors="ignore")
+    predictions = model.predict(X)
+    dataset["predicted_price"] = predictions
+    dataset.to_csv(PREDICTIONS_FILE, index=False)
+    generate_dashboard(dataset)
+    print("Predictions saved to 'predicted_prices.csv'.")
+
+# Generate a dashboard for visualization
+def generate_dashboard(data):
+    plt.figure(figsize=(10, 6))
+    plt.plot(data["time"], data["predicted_price"], label="Predicted Price", linestyle="--")
+    if "SEK_per_kWh" in data.columns:
+        plt.plot(data["time"], data["SEK_per_kWh"], label="Actual Price")
+    plt.xlabel("Time")
+    plt.ylabel("Electricity Price (SEK/kWh)")
+    plt.title("Electricity Prices: Predicted vs Actual")
+    plt.legend()
+    plt.grid()
+    plt.savefig("dashboard.png")
+    plt.close()
+
+# Schedule daily updates
+scheduler.add_job(predict_next_day_price, "cron", hour=23, minute=59)
+
+# API: Get predictions
+@app.get("/predictions")
+def get_predictions():
+    if not os.path.exists(PREDICTIONS_FILE):
+        return {"error": "Predictions not available"}
+    predictions = pd.read_csv(PREDICTIONS_FILE)
+    return predictions.to_dict()
+
+# API: Get dashboard
+@app.get("/dashboard")
+def get_dashboard():
+    if not os.path.exists("dashboard.png"):
+        return {"error": "Dashboard not available"}
+    return {"dashboard_url": "/dashboard.png"}