Machine Learning Stock Price Prediction Project

"""Demonstrate leakage-aware next-day price modelling on CSV or synthetic data."""

from __future__ import annotations

import argparse
from pathlib import Path

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_absolute_error


def demo_prices(rows: int = 700, random_state: int = 42) -> pd.DataFrame:
    rng = np.random.default_rng(random_state)
    dates = pd.date_range("2023-01-01", periods=rows, freq="B")
    returns = rng.normal(0.0003, 0.012, rows)
    close = 100 * np.exp(np.cumsum(returns))
    return pd.DataFrame({"Date": dates, "Close": close})


def load_prices(csv_path: Path | None) -> pd.DataFrame:
    frame = demo_prices() if csv_path is None else pd.read_csv(csv_path)
    required = {"Date", "Close"}
    if not required <= set(frame.columns):
        raise ValueError("CSV must contain Date and Close columns.")
    frame = frame.loc[:, ["Date", "Close"]].copy()
    frame["Date"] = pd.to_datetime(frame["Date"], errors="coerce")
    frame["Close"] = pd.to_numeric(frame["Close"], errors="coerce")
    frame = frame.dropna().sort_values("Date").drop_duplicates("Date")
    if len(frame) < 80 or (frame["Close"] <= 0).any():
        raise ValueError("Provide at least 80 dated rows with positive closing prices.")
    return frame


def make_features(prices: pd.DataFrame) -> pd.DataFrame:
    data = prices.copy()
    for lag in (1, 2, 3, 5, 10):
        data[f"lag_{lag}"] = data["Close"].shift(lag)
    data["mean_5"] = data["Close"].shift(1).rolling(5).mean()
    data["mean_20"] = data["Close"].shift(1).rolling(20).mean()
    data["target"] = data["Close"]
    return data.dropna().reset_index(drop=True)


def evaluate(prices: pd.DataFrame, random_state: int = 42) -> dict[str, float]:
    data = make_features(prices)
    feature_columns = [column for column in data.columns if column.startswith("lag_") or column.startswith("mean_")]
    split = int(len(data) * 0.8)
    train, test = data.iloc[:split], data.iloc[split:]
    model = RandomForestRegressor(n_estimators=200, min_samples_leaf=3, random_state=random_state, n_jobs=-1)
    model.fit(train[feature_columns], train["target"])
    predictions = model.predict(test[feature_columns])
    baseline = test["lag_1"]
    return {
        "model_mae": mean_absolute_error(test["target"], predictions),
        "baseline_mae": mean_absolute_error(test["target"], baseline),
        "test_rows": float(len(test)),
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Educational next-day closing-price model.")
    parser.add_argument("--csv", type=Path, help="Optional CSV with Date and Close columns")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    try:
        prices = load_prices(args.csv)
        metrics = evaluate(prices)
    except (ValueError, OSError) as error:
        raise SystemExit(f"Could not evaluate data: {error}") from error
    source = str(args.csv) if args.csv else "seeded synthetic random-walk data"
    print(f"Source: {source}")
    print(f"Time-ordered test rows: {int(metrics['test_rows'])}")
    print(f"Random forest MAE: {metrics['model_mae']:.4f}")
    print(f"Previous-close baseline MAE: {metrics['baseline_mae']:.4f}")
    print("Education only—not a forecast, trading signal, or investment recommendation.")


if __name__ == "__main__":
    main()