|
|
import yfinance as yf |
|
|
from prophet import Prophet |
|
|
import pandas as pd |
|
|
from typing import Dict, Any |
|
|
|
|
|
def generate_forecast(ticker: str) -> Dict[str, Any]: |
|
|
print(f"Generating forecast for ticker {ticker}...") |
|
|
|
|
|
stock_data = yf.download(ticker, period="2y", progress=False) |
|
|
|
|
|
if stock_data.empty: |
|
|
return {"error": f"Could not download historical data for {ticker}."} |
|
|
|
|
|
df_prophet = stock_data[['Close']].copy() |
|
|
df_prophet.reset_index(inplace=True) |
|
|
|
|
|
df_prophet.columns = ['ds', 'y'] |
|
|
|
|
|
model = Prophet( |
|
|
daily_seasonality=False, |
|
|
weekly_seasonality=True, |
|
|
yearly_seasonality=True, |
|
|
changepoint_prior_scale=0.05 |
|
|
) |
|
|
model.fit(df_prophet) |
|
|
|
|
|
future = model.make_future_dataframe(periods=30) |
|
|
forecast = model.predict(future) |
|
|
|
|
|
current_price = df_prophet['y'].iloc[-1] |
|
|
predicted_price_30_days = forecast['yhat'].iloc[-1] |
|
|
trend = "upward" if predicted_price_30_days > current_price else "downward" |
|
|
change_percent = ((predicted_price_30_days - current_price) / current_price) * 100 |
|
|
|
|
|
forecast_data = { |
|
|
"summary": ( |
|
|
f"The model predicts a {trend} trend over the next 30 days. " |
|
|
f"Current price: {current_price:.2f}, " |
|
|
f"predicted price in 30 days: {predicted_price_30_days:.2f} " |
|
|
f"({change_percent:+.2f}% change)." |
|
|
), |
|
|
|
|
|
"history_plot_data": [ |
|
|
{'ds': r['ds'].isoformat(), 'y': r['y']} for r in df_prophet.tail(90).to_dict('records') |
|
|
], |
|
|
"forecast_plot_data": [ |
|
|
{ |
|
|
'ds': r['ds'].isoformat(), |
|
|
'yhat': r['yhat'], |
|
|
'yhat_lower': r['yhat_lower'], |
|
|
'yhat_upper': r['yhat_upper'] |
|
|
} for r in forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(120).to_dict('records') |
|
|
] |
|
|
} |
|
|
|
|
|
print(f"Forecast for {ticker} generated successfully.") |
|
|
return forecast_data |
