Automated Web Scraping and Data Visualization with Python and AI — Part 6: Integrating AI Models for Predictive Analytics and In

Automated Web Scraping and Data Visualization with Python and AI — Part 6: Integrating AI Models for Predictive Analytics and Insights

Based on my technical understanding as a Lead Programmer Analyst (PHP, Perl, Python, Shell) and on the latest 2026 AI‑driven data‑pipeline trends, this deep‑dive shows you how to turn the raw tables you harvested in the earlier parts into forward‑looking predictions and visual stories.

Quick recap (Parts 1‑5): We started by designing a robust scraper (BeautifulSoup + Playwright), added error‑handling and rotating proxies (Part 2), stored the results in PostgreSQL (Part 3), visualized static charts with Matplotlib and Seaborn (Part 4), and finally wrapped everything into a Docker‑compose stack with CI/CD (Part 5). Now we close the loop by feeding that clean data into AI models, extracting insights, and visualizing the forecasts.

Why AI‑Powered Predictive Analytics?

• Speed. Modern LLM‑backed feature engineers (Claude 3.5 Sonnet, GPT‑4.5 Turbo) can suggest transformations in milliseconds, cutting the traditional data‑science “feature‑selection” phase by up to 70 %.
• Adaptability. With parallel‑agent architectures you can spin up a dedicated “trend‑detector” agent, a “seasonality‑model” agent, and a “business‑rule” agent, all communicating via lightweight websockets or Redis streams.
• Explainability. Prompt‑driven LLMs can generate natural‑language explanations of model outputs, turning a black‑box forecast into a stakeholder‑ready narrative.

Architecture Overview

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

Step 1 – Pull the Cleaned Dataset

All previous parts stored the final tidy DataFrame in a PostgreSQL table called public.sales_daily. The following snippet uses SQLAlchemy and pandas to load the data into memory.

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())

Step 2 – Let Claude 3.5 Sonnet Suggest Features

Claude’s Agentic Workflow lets you feed a prompt, get a JSON‑structured plan, and then execute it automatically. We’ll use the anthropic Python SDK (v0.6.0) to spin up a feature‑engineer agent that proposes transformations such as lag features, rolling averages, and holiday flags.

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))

Typical output (truncated for brevity):

[
  {"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"}
]

Step 3 – Execute the Feature Plan

We translate the JSON plan into concrete pandas operations. This block is deliberately modular so you can run it inside a Docker container that the parallel‑agent scheduler spawns.

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())

Step 4 – Parallel Model Training with GPT‑4.5 Turbo

OpenAI’s Turbo Parallel Agents (released Q2 2026) let you dispatch several model‑training jobs simultaneously and then aggregate results. In our example we’ll train three lightweight models:

• Prophet (seasonality‑aware)
• LightGBM (gradient‑boosted trees)
• Temporal Fusion Transformer (TFT) via PyTorch

Each model runs in its own container; a tiny orchestration script collects the MAE (Mean Absolute Error) and selects the best performer.

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']}")

Below are the minimal training scripts for each agent. They all read the engineered CSV from a shared volume (/data/engineered.csv) and write metrics.json back to the same folder.

Prophet Trainer (models/prophet/train.py)

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))

LightGBM Trainer (models/lightgbm/train.py)

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))

Temporal Fusion Transformer (TFT) Trainer (models/tft/train.py)

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))

Step 5 – Generate AI‑Driven Insights

Once the best model is identified (let’s say LightGBM), we can ask GPT‑4.5 Turbo to translate the raw forecast into a concise business narrative. The function‑calling feature ensures the output follows a predefined JSON schema, which our dashboard can consume directly.

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))

Sample output:

{
  "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."
  ]
}

Step 6 – Interactive Dashboard with Plotly Dash

Finally, we embed the forecast and AI‑generated narrative into a lightweight Dash app. The dashboard pulls the latest CSV, renders a line chart with confidence bands, and shows the natural‑language summary beneath.

import dash
from dash import dcc, html, Input, Output
import plotly.express as px
import pandas as pd
import json

app = dash.Dash(name)

Load data once (in production you’d cache or stream)

forecast = pd.read_csv("/models/lightgbm/

---

Originally published at https://artificial-inteligence.phptutorial.co.in