Price Monitoring Competitor Intelligence Workflows

Scheduling Price Reports with Python

A daily competitor-price report is only useful if it lands in the same inbox at the same time every morning, is identical whether it runs once or is retried after a blip, and can be regenerated for last Tuesday when someone asks. This recipe builds that job end to end: assemble the day’s summary, render it to HTML, CSV, and a plot, and deliver it on a schedule. It sits under automated price reporting and dashboards, which defines the reporting data model this job renders, and it consumes the index methodology from building a competitive price index. The emphasis here is operational: idempotency, backfill, timezone-safe day boundaries, and delivery that fails loudly rather than silently.

Prerequisites & Input Contract

The job reads from a materialized daily_summary table — one row per (day, category) with the four core KPIs — produced by the aggregation job described in the parent guide. This recipe does not recompute those metrics; it reads them, renders them, and ships them, which is exactly what keeps the emailed number identical to the dashboard number.

# Row contract from the daily_summary table.
# day is a date in a single declared timezone (UTC here).
summary_row = {
    "day": "2026-07-15",     # str ISO date, the reporting day
    "category": "audio",     # segment
    "price_index": 0.982,    # our price / median competitor (1.0 = parity)
    "win_rate": 0.612,       # share of assortment we lead on price
    "gross_margin": 0.284,
    "coverage": 0.874,       # observations / assortment
    "n_products": 412,
}

Library versions: Python 3.11+, pandas>=2.1, SQLAlchemy>=2.0, matplotlib>=3.8, and the standard-library smtplib/email for delivery. Install with pip install pandas sqlalchemy matplotlib. The scheduler options — cron, GitHub Actions, or APScheduler — are covered in the final step; the report logic itself is scheduler-agnostic.

Step-by-Step Implementation

Step 1 — Resolve the reporting day in one timezone

The most common scheduling bug is an ambiguous “today”. Compute the target day explicitly in UTC so the job produces the same slice regardless of which machine or region runs it. Accepting an optional override day is what makes backfill trivial later.

from datetime import datetime, timedelta, timezone

def reporting_day(override: str | None = None) -> str:
    """Yesterday in UTC by default; an explicit YYYY-MM-DD enables backfill."""
    if override:
        datetime.strptime(override, "%Y-%m-%d")   # validate format, raises if bad
        return override
    # A 06:00 job reports on the day that just fully completed: yesterday UTC.
    return (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y-%m-%d")

print(reporting_day())              # -> '2026-07-14' when run on 2026-07-15 UTC
print(reporting_day("2026-06-01"))  # -> '2026-06-01'  (backfill)

Step 2 — Assemble the summary and guard coverage

Pull the day’s rows plus 30 days of history for the trend chart. Before rendering anything, check coverage: a report over thin data misleads more than a skipped one, so a low-coverage day gets a banner rather than a confident-looking table.

import pandas as pd
from sqlalchemy import create_engine, text

def load_summary(engine, day: str) -> tuple[pd.DataFrame, pd.DataFrame, bool]:
    """Return (today_rows, 30d_history, coverage_ok)."""
    today = pd.read_sql(
        text("SELECT * FROM daily_summary WHERE day = :d"),
        engine, params={"d": day})
    if today.empty:
        raise RuntimeError(f"no summary rows for {day} — did aggregation run?")
    history = pd.read_sql(
        text("SELECT day, category, price_index FROM daily_summary "
             "WHERE day >= :start ORDER BY day"),
        engine, params={"start":
            (pd.to_datetime(day) - pd.Timedelta(days=30)).strftime("%Y-%m-%d")})
    coverage_ok = bool(today["coverage"].min() >= 0.60)   # coverage floor
    return today, history, coverage_ok

Step 3 — Render HTML, CSV, and a trend plot

Render the three artifacts from the loaded frame. The CSV is the machine-readable copy for anyone who wants the raw numbers; the HTML is the human view; the plot carries the trend a single day’s table cannot show. Matplotlib runs headless under the Agg backend so it works in a server job with no display.

import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from io import BytesIO

def render(today: pd.DataFrame, history: pd.DataFrame,
           day: str, coverage_ok: bool) -> dict:
    banner = ("" if coverage_ok else
              "<p style='color:#b45309;font-weight:700'>"
              "LOW COVERAGE — figures cover a partial assortment.</p>")
    table_html = (today.style
        .format({"price_index": "{:.3f}", "win_rate": "{:.1%}",
                 "gross_margin": "{:.1%}", "coverage": "{:.1%}"})
        .hide(axis="index").to_html())
    html = f"<h2>Competitive price report — {day}</h2>{banner}{table_html}"

    fig, ax = plt.subplots(figsize=(7, 3))
    for cat, grp in history.groupby("category"):
        ax.plot(grp["day"], grp["price_index"], marker="o", label=cat)
    ax.axhline(1.0, color="#6b7280", ls="--", lw=1)     # parity line
    ax.set_ylabel("price index"); ax.legend(fontsize=8); fig.autofmt_xdate()
    buf = BytesIO(); fig.savefig(buf, format="png", dpi=120, bbox_inches="tight")
    plt.close(fig)

    csv = today.to_csv(index=False).encode()
    return {"html": html, "csv": csv, "chart": buf.getvalue()}

Step 4 — Deliver with retry, then orchestrate

Delivery is the step most likely to fail transiently — a dropped SMTP connection, a rate-limited webhook — so wrap it in bounded retry with backoff. If every attempt fails, raise: the orchestrator must exit non-zero so the scheduler marks the run failed instead of pretending it delivered.

import os, ssl, smtplib, time
from email.message import EmailMessage

def deliver(artifacts: dict, to: list[str], day: str, retries: int = 3) -> None:
    msg = EmailMessage()
    msg["Subject"] = f"Competitive price report — {day}"
    msg["From"], msg["To"] = "pricing-bot@example.com", ", ".join(to)
    msg.set_content("Enable HTML to view the report inline.")
    msg.add_alternative(artifacts["html"], subtype="html")
    msg.add_attachment(artifacts["csv"], maintype="text", subtype="csv",
                       filename=f"summary_{day}.csv")
    msg.add_attachment(artifacts["chart"], maintype="image", subtype="png",
                       filename=f"trend_{day}.png")
    ctx = ssl.create_default_context()
    for attempt in range(1, retries + 1):
        try:
            with smtplib.SMTP_SSL("smtp.example.com", 465, context=ctx) as s:
                s.login("pricing-bot@example.com", os.environ["SMTP_PASSWORD"])
                s.send_message(msg)
            return                                    # delivered, done
        except (smtplib.SMTPException, OSError) as e:
            if attempt == retries:
                raise RuntimeError(f"delivery failed after {retries} tries: {e}")
            time.sleep(2 ** attempt)                  # 2s, 4s backoff

def run(day_override: str | None = None) -> None:
    engine = create_engine(os.environ["PRICE_DB_URL"])
    day = reporting_day(day_override)
    today, history, coverage_ok = load_summary(engine, day)
    artifacts = render(today, history, day, coverage_ok)
    deliver(artifacts, to=["pricing@example.com"], day=day)
    print(f"report for {day} delivered ({len(today)} segments)")

if __name__ == "__main__":
    import sys
    run(sys.argv[1] if len(sys.argv) > 1 else None)   # optional backfill day

Running the job for a normal day prints:

report for 2026-07-14 delivered (2 segments)

Verification & Testing

Test the day resolution and delivery-retry logic without sending real email. The retry test uses a stub that fails twice then succeeds, proving the backoff loop recovers rather than aborts.

def test_reporting_day():
    assert reporting_day("2026-06-01") == "2026-06-01"
    # Default is a valid ISO date, exactly one day behind UTC now.
    import re
    assert re.fullmatch(r"\d{4}-\d{2}-\d{2}", reporting_day())

def test_delivery_retries(monkeypatch):
    calls = {"n": 0}
    def flaky_send(self, msg):
        calls["n"] += 1
        if calls["n"] < 3:
            raise smtplib.SMTPServerDisconnected("transient")
    monkeypatch.setattr(smtplib.SMTP_SSL, "send_message", flaky_send)
    monkeypatch.setattr(smtplib.SMTP_SSL, "login", lambda *a, **k: None)
    deliver({"html": "<p>x</p>", "csv": b"a", "chart": b"p"},
            ["x@example.com"], "2026-07-14")
    assert calls["n"] == 3          # failed twice, third attempt delivered
    print("retry logic verified")

For an end-to-end check, run python report.py 2026-07-14 twice against a fixture database and confirm the two emails are byte-identical — that proves the render step is deterministic and safe to retry.

Edge Cases & Gotchas

  • Missing data day. If the aggregation job did not run, daily_summary has no rows and load_summary raises rather than emailing an empty table. That is deliberate: a hard failure that pages someone beats a blank report that looks like a real (and alarming) zero. Detecting why the data is missing is the job of price anomaly alerting and monitoring; this job’s duty is to refuse to ship nonsense.
  • Timezone and cutoff drift. A 06:00 local job that computes “today” from server-local time will silently shift its window when the host moves regions or when daylight-saving flips. Pin the reporting day to UTC (Step 1) and let the scheduler, not the code, decide wall-clock send time.
  • Delivery failure treated as success. A naive try/except: pass around send_message turns a bounced report into a green run. The retry loop re-raises on final failure precisely so the process exits non-zero and the scheduler surfaces it.
  • Backfill duplication. Because the report only reads the summary table, re-running for a past day is safe and produces the same artifacts — provided the upstream materialization used the idempotent upsert from the parent guide. If you regenerate the summary first, use delete-then-insert so backfill never doubles rows.

Performance Notes

The job is I/O-bound, not compute-bound: it reads a few thousand pre-aggregated rows, renders a small table and one figure, and sends one email — all comfortably under a second of CPU on a single worker. The expensive group-by already ran in the aggregation job, which is the whole point of materializing the summary. The only step that scales with history is the 30-day trend query; keep daily_summary indexed on day and it stays a millisecond range scan. If you fan out to many per-category reports, render once and slice the frame per recipient rather than re-querying the database for each; the database round-trip, not the rendering, dominates at that point.

Scheduling Options

Three schedulers cover the spectrum. Cron is the simplest — 0 6 * * * cd /app && python report.py — appropriate when you already own a reliable host and are content managing secrets and log rotation yourself. GitHub Actions (a schedule: workflow) is the strongest default for most teams: run history is visible, secrets live in the platform store, and a failed run notifies through the normal build path, giving you CI-driven reporting for free. APScheduler fits when the report must run inside a long-lived Python service alongside other jobs and you want in-process scheduling without an external cron. Whichever you pick, the report code is identical — it takes an optional day argument and exits non-zero on failure, so the scheduler only needs to invoke it and watch the exit code.

Frequently Asked Questions

Should the report job recompute metrics or just read them? Just read them. Recomputing in the report is how the emailed number drifts from the dashboard number. Compute each KPI once in the aggregation job, materialize it, and let this job render the materialized rows unchanged.

How do I backfill a report for a past day? Pass the date as an argument: python report.py 2026-06-01. The job resolves that exact day instead of yesterday, reads the existing summary rows, and re-renders. Because it only reads, backfilling is idempotent and safe to repeat.

What should happen when a day has no data? Fail loudly. The job raises when daily_summary has no rows for the target day, so the scheduler marks the run failed and someone investigates. Emailing an empty or zero-filled table hides the outage and can look like a real, alarming result.