Component
Technology (2026)
Responsibility


Web Scraper
Playwright + BrightData GPT‑Vision (image‑aware extraction)
Collect raw HTML/visual tables from target sites


Data Lake
PostgreSQL + MinIO (S3‑compatible)
Persist raw and cleaned data for downstream jobs


Feature‑Engineering Agent
Claude 3.5 Sonnet (Agentic Workflow)
Suggest & execute feature transformations via Python scripts


Model Training Agent
GPT‑4.5 Turbo (parallel inference)
Run LightGBM, Prophet, or PyTorch models in parallel containers


Insight Generator
OpenAI Function‑Calling + LangChain
Translate model predictions into human‑readable insights


Visualization Layer
Plotly Dash + Streamlit
Interactive dashboards for business users

import os
import pandas as pd
from sqlalchemy import create_engine
from dotenv import load_dotenv

load_dotenv()  # .env contains DB credentials

DB_URL = (
    f"postgresql+psycopg2://{os.getenv('DB_USER')}:"
    f"{os.getenv('DB_PASS')}@{os.getenv('DB_HOST')}:"
    f"{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
)

engine = create_engine(DB_URL)
query = """
    SELECT
        date,
        product_id,
        region,
        units_sold,
        revenue,
        discount_pct
    FROM public.sales_daily
    WHERE date >= CURRENT_DATE - INTERVAL '180 days'
    ORDER BY date;
"""
df = pd.read_sql(query, engine)
df['date'] = pd.to_datetime(df['date'])
print(df.head())

import anthropic
import json
from pathlib import Path

client = anthropic.Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))

prompt = f"""
You are a data‑science assistant. The DataFrame `df` has columns:
- date (datetime)
- product_id (int)
- region (str)
- units_sold (int)
- revenue (float)
- discount_pct (float)

Suggest a JSON list of feature engineering steps that would improve a time‑series forecast of `units_sold`. Include:
1. Lag features (e.g., units_sold_lag_7)
2. Rolling statistics (e.g., revenue_rolling_14)
3. Categorical encodings (e.g., region_one_hot)
4. Holiday flag (use US federal holidays)

Return ONLY the JSON, no prose.
"""

response = client.messages.create(
    model="claude-3-5-sonnet-20240620",
    max_tokens=800,
    temperature=0.0,
    messages=[{"role": "user", "content": prompt}]
)

# Extract JSON from the response
raw_json = response.content[0].text.strip()
features_plan = json.loads(raw_json)
print(json.dumps(features_plan, indent=2))

[
  {"type":"lag","column":"units_sold","lag":7},
  {"type":"lag","column":"units_sold","lag":30},
  {"type":"rolling","column":"revenue","window":14,"agg":"mean"},
  {"type":"rolling","column":"discount_pct","window":7,"agg":"max"},
  {"type":"one_hot","column":"region"},
  {"type":"holiday_flag","country":"US","column":"date"}
]

import pandas as pd
import holidays
from sklearn.preprocessing import OneHotEncoder

def apply_feature_plan(df: pd.DataFrame, plan: list) -> pd.DataFrame:
    df = df.copy()
    for step in plan:
        if step["type"] == "lag":
            lag = step["lag"]
            col = step["column"]
            df[f"{col}_lag_{lag}"] = df.groupby('product_id')[col].shift(lag)
        elif step["type"] == "rolling":
            win = step["window"]
            col = step["column"]
            agg = step["agg"]
            rolled = df.groupby('product_id')[col].transform(
                lambda s: s.rolling(window=win, min_periods=1).agg(agg)
            )
            df[f"{col}_rolling_{win}"] = rolled
        elif step["type"] == "one_hot":
            encoder = OneHotEncoder(sparse=False, drop='first')
            ohe = encoder.fit_transform(df[[step["column"]]])
            ohe_cols = encoder.get_feature_names_out([step["column"]])
            df[ohe_cols] = ohe
            df.drop(columns=[step["column"]], inplace=True)
        elif step["type"] == "holiday_flag":
            us_holidays = holidays.US()
            df["is_holiday"] = df[step["column"]].apply(lambda d: d in us_holidays)
    return df

engineered_df = apply_feature_plan(df, features_plan)
print(engineered_df.head())

import subprocess, json, os, time
from pathlib import Path

# Directory layout:
# /models/prophet/train.py
# /models/lightgbm/train.py
# /models/tft/train.py
BASE_DIR = Path(__file__).parent

def launch_agent(script_path: Path, model_name: str):
    """Spawn a subprocess that runs the training script.
    The script must write a JSON file `metrics.json` to its cwd."""
    env = os.environ.copy()
    env["MODEL_NAME"] = model_name
    proc = subprocess.Popen(
        ["python", str(script_path)],
        cwd=script_path.parent,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True
    )
    return proc

agents = {
    "prophet": launch_agent(BASE_DIR / "models/prophet/train.py", "prophet"),
    "lightgbm": launch_agent(BASE_DIR / "models/lightgbm/train.py", "lightgbm"),
    "tft": launch_agent(BASE_DIR / "models/tft/train.py", "tft")
}

# Simple poll loop (in production you'd use Celery/Redis)
results = {}
while agents:
    for name, proc in list(agents.items()):
        if proc.poll() is not None:  # finished
            out, _ = proc.communicate()
            print(f"--- {name} output ---")
            print(out)
            metrics_path = proc.cwd / "metrics.json"
            if metrics_path.exists():
                results[name] = json.load(metrics_path.open())
            else:
                results[name] = {"status":"failed"}
            del agents[name]
    time.sleep(2)

# Pick the best model based on MAE
best_model = min(results, key=lambda k: results[k].get("mae", float("inf")))
print(f"Best model: {best_model} with MAE={results[best_model]['mae']}")

import pandas as pd
from prophet import Prophet
import json
from pathlib import Path

DATA_PATH = Path("/data/engineered.csv")
df = pd.read_csv(DATA_PATH, parse_dates=["date"])

# Prophet expects columns ds (date) and y (target)
prophet_df = df.rename(columns={"date":"ds", "units_sold":"y"})[["ds","y"]]

model = Prophet(yearly_seasonality=True, weekly_seasonality=True, daily_seasonality=False)
model.fit(prophet_df)

future = model.make_future_dataframe(periods=30)
forecast = model.predict(future)

# Simple MAE on the validation slice (last 30 days of historic)
val = forecast.tail(30)
actual = df.tail(30)["units_sold"].reset_index(drop=True)
mae = (val["yhat"].reset_index(drop=True) - actual).abs().mean()

metrics = {"mae": mae, "model":"prophet"}
Path("metrics.json").write_text(json.dumps(metrics))

import pandas as pd
import lightgbm as lgb
import json
from pathlib import Path
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error

df = pd.read_csv("/data/engineered.csv", parse_dates=["date"])
X = df.drop(columns=["units_sold", "date"])
y = df["units_sold"]

X_train, X_val, y_train, y_val = train_test_split(
    X, y, test_size=0.2, shuffle=False  # preserve temporal order
)

train_set = lgb.Dataset(X_train, label=y_train)
val_set = lgb.Dataset(X_val, label=y_val, reference=train_set)

params = {
    "objective": "regression",
    "metric": "mae",
    "learning_rate": 0.05,
    "num_leaves": 31,
    "verbosity": -1,
}

gbm = lgb.train(params, train_set, valid_sets=[val_set], early_stopping_rounds=30, verbose_eval=False)

preds = gbm.predict(X_val, num_iteration=gbm.best_iteration)
mae = mean_absolute_error(y_val, preds)

metrics = {"mae": mae, "model":"lightgbm"}
Path("metrics.json").write_text(json.dumps(metrics))

import pandas as pd
import torch
import json
from pathlib import Path
from pytorch_forecasting import TimeSeriesDataSet, TemporalFusionTransformer, Baseline
from pytorch_forecasting.metrics import MAE

df = pd.read_csv("/data/engineered.csv", parse_dates=["date"])
df["time_idx"] = (df["date"] - df["date"].min()).dt.days

max_encoder_length = 30
max_prediction_length = 7

training = TimeSeriesDataSet(
    df,
    time_idx="time_idx",
    target="units_sold",
    group_ids=["product_id"],
    max_encoder_length=max_encoder_length,
    max_prediction_length=max_prediction_length,
    static_categoricals=["region"],
    static_reals=[],
    time_varying_known_categoricals=[],
    time_varying_known_reals=["revenue", "discount_pct", "is_holiday"],
    time_varying_unknown_categoricals=[],
    time_varying_unknown_reals=["units_sold"]
)

val = TimeSeriesDataSet.from_dataset(training, df, predict=True, stop_randomization=True)

train_loader = torch.utils.data.DataLoader(training, batch_size=64, shuffle=True)
val_loader = torch.utils.data.DataLoader(val, batch_size=64, shuffle=False)

tft = TemporalFusionTransformer.from_dataset(
    training,
    learning_rate=0.03,
    hidden_size=16,
    attention_head_size=1,
    dropout=0.1,
    loss=MAE(),
    optimizer="adam",
    reduce_on_plateau_patience=4,
)

trainer = torch.optim.Adam(tft.parameters(), lr=0.03)
epochs = 8
for epoch in range(epochs):
    tft.fit(train_loader, val_loader, max_epochs=1, verbose=False)

# Validation MAE
actuals = torch.cat([y for x, y in iter(val_loader)])
predictions = tft.predict(val_loader)
mae = MAE()(predictions, actuals).item()

metrics = {"mae": mae, "model":"tft"}
Path("metrics.json").write_text(json.dumps(metrics))

import openai
import os
import json

openai.api_key = os.getenv("OPENAI_API_KEY")

# Load the 30‑day forecast produced by the best model
forecast_path = Path(f"/models/{best_model}/forecast.csv")
forecast_df = pd.read_csv(forecast_path)

prompt = f"""
You are an analytics consultant. Using the following forecast for `units_sold` (next 30 days) and the historical context (average daily sales = {df['units_sold'].mean():.1f}), write a short (max 150 words) executive summary that includes:

1. Expected growth/decline trend.
2. Any dates where sales exceed the 95th percentile of historic daily sales.
3. Recommendations for inventory or promotion actions.

Return the answer in JSON with fields: `summary`, `high_volume_dates`, `recommendations`.
"""

response = openai.ChatCompletion.create(
    model="gpt-4.5-turbo-202406",
    messages=[{"role":"user","content":prompt}],
    temperature=0.2,
    max_tokens=300,
    functions=[{
        "name":"insight_schema",
        "description":"Schema for predictive insight",
        "parameters":{
            "type":"object",
            "properties":{
                "summary":{"type":"string"},
                "high_volume_dates":{"type":"array","items":{"type":"string"}},
                "recommendations":{"type":"array","items":{"type":"string"}}
            },
            "required":["summary","high_volume_dates","recommendations"]
        }
    }],
    function_call={"name":"insight_schema"}
)

insight = response.choices[0].message.function_call.arguments
insight_json = json.loads(insight)
print(json.dumps(insight_json, indent=2))

{
  "summary":"The model predicts a modest 3 % upward trend in daily units sold over the next month, driven primarily by the upcoming promotional week (June 12‑18). Sales are expected to peak on June 15, crossing the historic 95th‑percentile threshold.",
  "high_volume_dates":["2026-06-15"],
  "recommendations":[
    "Increase inventory for product #42 in the West region before June 14.",
    "Launch a targeted discount on June 12‑14 to smooth demand spikes."
  ]
}