maskantech/maskanx_cm_backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(campaigns): enforce budget guardrails automatically

Browse Source 
Adds the insights reader, a pure guardrail evaluator, and the sweep that
applies them to live campaigns.

Three things carry the weight here:

Units. Meta reports spend in major units ("12.34") while budgets and
guardrails are in minor units (1234). normalise_row converts, rounding
half up rather than using round(), which rounds halves to even and can
record an exact half-unit of spend as nothing.

Leads. Meta has no leads field; leads live in the actions array, under
several action types depending on whether the lead came from a Facebook
form or a pixel. Cost per lead is computed from spend and leads over the
same window rather than read from cost_per_action_type, so the two can
never disagree.

Order. The campaign is paused on Meta before the local record changes. A
campaign recorded as paused but still delivering is the outcome this
exists to prevent. One campaign's failure never aborts the sweep, so a
rate limit on the third does not leave the fourth unguarded.

Cost rules are skipped until the first lead or click: no leads yet is not
an infinite cost per lead, and pausing for that would kill every campaign
in its first hour.

Deliberately a deterministic loop rather than a maskanx_cron_jobs entry.
That scheduler runs prompts through an agent, and asking a language model
whether a budget has been exceeded would make an arithmetic guarantee
probabilistic. Follows reconcile.py's lifespan-task pattern instead, and
warns every cycle if Meta is unconfigured — a safety system that cannot
run should be loud.

require_approval_for_budget_increase is enforced at the API layer, where
the budget is actually edited: raising it while a campaign awaits approval
returns 409, since it would change what the approver is reviewing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
AFFAANh
2026-08-03 18:04:25 +05:30
co-authored by Claude Opus 5
parent e8abe33d18
commit 6c39d42874
11 changed files with 1385 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+42
View File
@@ -192,6 +192,48 @@ Campaigns that have not reached Meta yet (`draft`, `pending_approval`,
The interval defaults to 120 seconds and is set with
`MASKANX_CAMPAIGN_RECONCILE_SECONDS`; `0` disables the loop.
### Guardrails
Guardrails are the only thing in MaskanX that halts spending without a
human. A background sweep reads each **live** campaign's spend from Meta and
applies the campaign's own rules:
| Rule | Window | Action |
| --- | --- | --- |
| `max_campaign_spend` | lifetime | stop |
| `auto_pause_if_spend_reaches` | today | pause |
| `max_cost_per_lead` | today | pause |
| `max_cost_per_click` | today | pause |
The most severe breach wins, so a campaign over both its lifetime cap and
its cost per lead is stopped rather than merely paused. `stop_loss_enabled`
gates only `max_campaign_spend`, the one rule that ends a campaign
permanently; turning it off leaves the pause rules working.
Every amount is in **minor** currency units (paise, cents), matching the
budget fields. Cost rules are skipped until the campaign has its first lead
or click — no leads yet is not an infinite cost per lead, and pausing for
that would kill every campaign in its first hour.
The campaign is paused **on Meta first**, then recorded locally: a campaign
recorded as paused but still delivering is the outcome the whole mechanism
exists to prevent. Each action writes an event naming the rule and the
observed value, so a campaign that stopped overnight can be explained the
next morning, and the Campaigns page tags it.
Interval: `MASKANX_GUARDRAIL_INTERVAL_SECONDS`, default 900 (15 minutes).
`0` disables enforcement, which is logged as a warning at startup — a safety
system that is not running should be loud.
This is a deterministic loop, **not** a `maskanx_cron_jobs` entry. That
scheduler runs prompts through an agent, and asking a language model whether
a budget has been exceeded would make an arithmetic guarantee probabilistic.
`require_approval_for_budget_increase` is enforced at the API layer instead:
raising a budget on a campaign that is awaiting approval returns `409`,
because it would change the figure the approver is reviewing. Reject it back
to draft, edit, and resubmit.
### Live smoke test
The unit suite runs entirely against a fake transport: it proves the code
src/adclaw/app/_app.py
+23
View File
@@ -282,6 +282,27 @@ async def lifespan(app: FastAPI): # pylint: disable=too-many-statements
"META_ADS_ACCOUNT_ID not set; campaign reconciliation not started",
)
# --- Guardrail enforcement ---
# Started regardless of META_ADS_ACCOUNT_ID: it finds its work by listing
# live campaigns, not by scanning an account. A missing Meta token is
# reported by the loop each cycle rather than silently skipped here — a
# safety system that is not running should be loud.
guardrail_task = None
from ..campaigns.enforce import enforce_loop, guardrail_interval_seconds
if guardrail_interval_seconds() > 0:
from ..campaigns.repo import CampaignRepository
from ..meta.client import MetaClient, access_token_from_env
def _guardrail_meta_client():
return MetaClient(access_token=access_token_from_env())
guardrail_task = asyncio.create_task(
enforce_loop(CampaignRepository, _guardrail_meta_client),
name="campaign_guardrails",
)
app.state.guardrail_task = guardrail_task
try:
if mcp_initial_config is not None:
mcp_init_task = _schedule_mcp_initialization(
@@ -295,6 +316,8 @@ async def lifespan(app: FastAPI): # pylint: disable=too-many-statements
app.state.watchdog.stop()
if reconcile_task is not None:
reconcile_task.cancel()
if guardrail_task is not None:
guardrail_task.cancel()
# stop order: watchers -> cron -> channels -> mcp -> runner
try:
await config_watcher.stop()
src/adclaw/app/routers/campaigns.py
+38
View File
@@ -261,6 +261,43 @@ async def list_campaign_events(campaign_id: str) -> list[dict[str, Any]]:
return await get_repository().list_events(campaign_id)
def _assert_budget_increase_allowed(
campaign: CampaignSpec,
new_budget: dict[str, Any],
) -> None:
"""Refuse to raise the budget out from under a pending approval.
With `require_approval_for_budget_increase` set, an approver reviewing a
campaign must be reviewing the number it will actually spend. Raising it
while the campaign sits in `pending_approval` would leave the approval
attached to a figure nobody agreed to.
Only increases are blocked; lowering a budget needs no re-approval, and
a campaign still in `draft` has no pending review to invalidate. Reject
it back to draft, change the budget, and submit again.
"""
if not campaign.guardrails.get("require_approval_for_budget_increase"):
return
if campaign.status != "pending_approval":
return
for field in ("daily_budget", "lifetime_budget"):
old = campaign.budget.get(field)
new = new_budget.get(field)
if old is None or new is None:
continue
if new > old:
raise HTTPException(
status_code=http_status.HTTP_409_CONFLICT,
detail=(
f"This campaign requires approval for a budget increase, "
f"and it is awaiting approval. Raising {field} from {old} "
f"to {new} would change what is being reviewed. Reject it "
f"back to draft first, then edit and resubmit."
),
)
@router.put("/{campaign_id}", response_model=CampaignSpec)
async def update_campaign(
campaign_id: str,
@@ -276,6 +313,7 @@ async def update_campaign(
effective_budget = updates.get("budget", campaign.budget)
effective_guardrails = updates.get("guardrails", campaign.guardrails)
effective_account = updates.get("ad_account_id", campaign.ad_account_id)
_assert_budget_increase_allowed(campaign, effective_budget)
await _validate_or_422(effective_budget, effective_guardrails, effective_account)
for field, value in updates.items():
setattr(campaign, field, value)
src/adclaw/campaigns/enforce.py
+195
View File
@@ -0,0 +1,195 @@
# -*- coding: utf-8 -*-
"""Apply guardrails to live campaigns, on a schedule.
This is the only part of MaskanX that halts spending without a human, so
its failure modes matter more than its features:
- It acts **only** on campaigns that are `live`. Nothing else can be
spending, and acting on anything else would fight the operator.
- It pauses on Meta **before** recording the change locally. A campaign
recorded as paused but still delivering is the outcome this whole module
exists to prevent.
- One campaign's failure never stops the others being checked. A rate limit
on campaign three must not leave campaigns four and five unguarded.
- Every action writes an event naming the rule and the observed value, so a
campaign that stopped overnight can be explained the next morning.
Deliberately not an LLM cron job. MaskanX's `maskanx_cron_jobs` scheduler
runs prompts through an agent; asking a language model whether a budget has
been exceeded would make an arithmetic guarantee probabilistic. This
follows `reconcile.py` instead: a deterministic loop wired into the app
lifespan.
"""
from __future__ import annotations
import asyncio
import logging
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from .guardrails import GuardrailBreach, evaluate
from .state import next_status
from ..meta.client import MetaNotConfiguredError
from ..meta.insights import total_metrics
logger = logging.getLogger(__name__)
INTERVAL_ENV = "MASKANX_GUARDRAIL_INTERVAL_SECONDS"
DEFAULT_INTERVAL_SECONDS = 900.0 # 15 minutes, per the design spec.
@dataclass
class EnforcementResult:
paused: list[str] = field(default_factory=list)
stopped: list[str] = field(default_factory=list)
checked: int = 0
failed: list[str] = field(default_factory=list)
@property
def acted(self) -> bool:
return bool(self.paused or self.stopped)
def guardrail_interval_seconds() -> float:
"""Return the check interval; 0 or less disables enforcement."""
raw = os.environ.get(INTERVAL_ENV)
if raw is None:
return DEFAULT_INTERVAL_SECONDS
try:
return max(0.0, float(raw))
except ValueError:
logger.warning(
"%s is not a number (%r); using the default of %.0fs",
INTERVAL_ENV,
raw,
DEFAULT_INTERVAL_SECONDS,
)
return DEFAULT_INTERVAL_SECONDS
async def observed_spend(meta, campaign) -> dict[str, int | None]:
"""Read today's and lifetime spend for one campaign.
Two calls rather than one: a guardrail on today's spend and a stop-loss
on total spend are different windows, and Meta will not return both in
a single row.
A campaign with no delivery yet returns zeroes rather than raising, so
the caller can evaluate it like any other.
"""
today_rows = await meta.get_insights(
campaign.meta_campaign_id, date_preset="today",
)
lifetime_rows = await meta.get_insights(
campaign.meta_campaign_id, date_preset="maximum",
)
today = total_metrics(today_rows)
lifetime = total_metrics(lifetime_rows)
return {
"today_spend": today["spend"],
"lifetime_spend": lifetime["spend"],
"cost_per_lead": today["cost_per_lead"],
"cost_per_click": today["cost_per_click"],
}
async def _apply(repo, meta, campaign, breach: GuardrailBreach) -> None:
"""Halt the campaign on Meta, then record it. Order matters."""
await meta.update_object_status(campaign.meta_campaign_id, "PAUSED")
campaign.status = next_status(campaign.status, breach.action)
campaign.advanced["last_guardrail_trip"] = {
"rule": breach.rule,
"action": breach.action,
"limit": breach.limit,
"observed": breach.observed,
"at": datetime.now(timezone.utc).isoformat(),
}
await repo.update_campaign_with_event(
campaign,
event_type=f"campaign.guardrail_{breach.action}",
actor="guardrail",
reason=breach.reason,
payload={
"rule": breach.rule,
"limit": breach.limit,
"observed": breach.observed,
},
)
logger.warning(
"Guardrail %s campaign %s: %s", breach.action, campaign.id, breach.reason,
)
async def enforce_once(repo, meta) -> EnforcementResult:
"""Check every live campaign once. Returns what was acted on."""
result = EnforcementResult()
campaigns = await repo.list_campaigns(status="live")
for campaign in campaigns:
if not campaign.meta_campaign_id:
# Live without a Meta object should not happen, but if it does
# there is nothing to read spend from and nothing to pause.
continue
try:
result.checked += 1
spend = await observed_spend(meta, campaign)
breach = evaluate(campaign.guardrails or {}, **spend)
if breach is None:
continue
await _apply(repo, meta, campaign, breach)
if breach.action == "stop":
result.stopped.append(campaign.id)
else:
result.paused.append(campaign.id)
except Exception as exc:
# One campaign failing must not leave the rest unguarded, so
# this catches broadly and continues rather than aborting the
# sweep. The campaign keeps running, which is why the failure
# is logged at warning rather than swallowed.
result.failed.append(campaign.id)
logger.warning(
"Guardrail check failed for campaign %s: %s", campaign.id, exc,
)
return result
async def enforce_loop(repo_factory, meta_factory) -> None:
"""Enforce forever. Never raises: a bad cycle is logged and retried."""
interval = guardrail_interval_seconds()
if interval <= 0:
logger.warning(
"Guardrail enforcement is DISABLED (%s=0). Campaigns will not be "
"paused automatically when they exceed their limits.",
INTERVAL_ENV,
)
return
logger.info("Guardrail enforcement started, every %.0fs", interval)
while True:
try:
await asyncio.sleep(interval)
result = await enforce_once(repo_factory(), meta_factory())
if result.acted:
logger.warning(
"Guardrails acted: %d paused, %d stopped (of %d checked)",
len(result.paused),
len(result.stopped),
result.checked,
)
except asyncio.CancelledError:
logger.info("Guardrail enforcement stopped")
raise
except MetaNotConfiguredError:
# Warned every cycle on purpose. A safety system that cannot run
# should be loud, not quietly absent.
logger.warning(
"Guardrails cannot run: Meta is not configured. Set "
"META_ADS_ACCESS_TOKEN in Settings > Environments. Campaigns "
"will not be paused automatically until then.",
)
except Exception:
logger.exception("Guardrail enforcement cycle failed")
src/adclaw/campaigns/guardrails.py
+127
View File
@@ -0,0 +1,127 @@
# -*- coding: utf-8 -*-
"""Decide whether a campaign's own rules say it should stop spending.
Pure evaluation: no I/O, no Meta calls, no database. `evaluate` takes the
guardrails an operator set and the spend actually observed, and returns the
action to take. Applying that action is `enforce.py`'s job.
Keeping the decision separate from the acting is what makes this testable
without a fake Meta, and it is worth the split because these rules are the
last automatic defence against a campaign spending more than intended.
Every amount here is in **minor** currency units (paise, cents), matching
`meta.insights.normalise_row` and Graph's budget fields. Mixing major and
minor units would make a limit either 100x too loose or 100x too tight.
Two rules deliberately do NOT live here:
- `require_approval_for_budget_increase` is enforced at the API layer, when
the budget is edited. There is nothing for a periodic check to do about a
budget that has already been raised.
- `daily_budget_limit` / `lifetime_budget_limit` are validated when the
budget is set (`validation.py`). They cap what may be configured; these
rules cap what may actually be spent.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class GuardrailBreach:
"""One rule that has been broken, and what to do about it."""
rule: str
action: str # "pause" or "stop"
limit: int
observed: int
@property
def reason(self) -> str:
return (
f"{self.rule}: observed {self.observed} against a limit of "
f"{self.limit} (minor currency units)"
)
def _limit(guardrails: dict[str, Any], key: str) -> int | None:
"""Read a numeric limit, treating anything unusable as 'not set'.
A limit of 0 is honoured as a real limit, not read as absent: someone
who sets a cap of zero means "do not spend", and `if not value` would
silently discard that.
"""
value = guardrails.get(key)
if value is None or isinstance(value, bool):
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def evaluate(
guardrails: dict[str, Any],
*,
today_spend: int,
lifetime_spend: int,
cost_per_lead: int | None = None,
cost_per_click: int | None = None,
) -> GuardrailBreach | None:
"""Return the breach that should act on this campaign, or None.
Rules are checked most-severe first and the first match wins, so a
campaign that has blown both its lifetime cap and its cost per lead is
stopped rather than merely paused. Returning one breach rather than a
list keeps the caller from having to decide which of several actions to
apply.
`stop_loss_enabled` gates only `max_campaign_spend`, the one rule that
ends a campaign permanently. Turning it off leaves the pause rules
working, which is the point of having it as a separate switch.
"""
lifetime_cap = _limit(guardrails, "max_campaign_spend")
if (
lifetime_cap is not None
and guardrails.get("stop_loss_enabled")
and lifetime_spend >= lifetime_cap
):
return GuardrailBreach(
rule="max_campaign_spend",
action="stop",
limit=lifetime_cap,
observed=lifetime_spend,
)
daily_cap = _limit(guardrails, "auto_pause_if_spend_reaches")
if daily_cap is not None and today_spend >= daily_cap:
return GuardrailBreach(
rule="auto_pause_if_spend_reaches",
action="pause",
limit=daily_cap,
observed=today_spend,
)
# Cost rules are skipped while the observed value is None: a campaign
# with no leads yet has no cost per lead, and pausing it for that would
# kill every campaign in its first hour.
cpl_cap = _limit(guardrails, "max_cost_per_lead")
if cpl_cap is not None and cost_per_lead is not None and cost_per_lead > cpl_cap:
return GuardrailBreach(
rule="max_cost_per_lead",
action="pause",
limit=cpl_cap,
observed=cost_per_lead,
)
cpc_cap = _limit(guardrails, "max_cost_per_click")
if cpc_cap is not None and cost_per_click is not None and cost_per_click > cpc_cap:
return GuardrailBreach(
rule="max_cost_per_click",
action="pause",
limit=cpc_cap,
observed=cost_per_click,
)
return None
src/adclaw/meta/client.py
+60
View File
@@ -52,6 +52,26 @@ CREATE_STATUS = "PAUSED"
LIST_FIELDS = "id,name,status,effective_status,created_time"
# Metrics pulled for every insights read. `actions` and `cost_per_action_type`
# are where leads live: Meta reports them as action rows, not as a top-level
# field, so anything lead-related has to be dug out of those arrays.
INSIGHT_FIELDS = ",".join(
(
"spend",
"impressions",
"reach",
"frequency",
"clicks",
"ctr",
"cpc",
"cpm",
"actions",
"cost_per_action_type",
"date_start",
"date_stop",
),
)
# Path suffixes that create a campaign/ad set/ad. _post refuses any of these
# with a non-PAUSED status as defence in depth, in case a future caller
# bypasses the typed create_* guards and calls _post directly.
@@ -203,6 +223,46 @@ class MetaClient:
{"fields": AD_ACCOUNT_FIELDS},
)
async def get_insights(
self,
object_id: str,
*,
date_preset: str | None = None,
since: str | None = None,
until: str | None = None,
level: str | None = None,
breakdowns: list[str] | None = None,
time_increment: int | str | None = None,
) -> list[dict[str, Any]]:
"""Return insight rows for a campaign, ad set, ad or ad account.
Pass either `date_preset` (e.g. "today", "yesterday") or an explicit
`since`/`until` pair of YYYY-MM-DD dates; passing neither gives
Meta's own default window, which is not what a caller measuring
spend wants, so one of the two should always be supplied.
Returns the raw rows. Meta reports every metric as a string and
reports leads inside the `actions` array rather than as a field of
its own, so callers should go through `adclaw.meta.insights` rather
than reading these dicts directly.
An object with no delivery in the window returns `[]`, not zeroes.
"""
params: dict[str, Any] = {"fields": INSIGHT_FIELDS}
if date_preset:
params["date_preset"] = date_preset
if since and until:
params["time_range"] = json.dumps({"since": since, "until": until})
if level:
params["level"] = level
if breakdowns:
params["breakdowns"] = ",".join(breakdowns)
if time_increment is not None:
params["time_increment"] = time_increment
result = await self._get(f"/{object_id}/insights", params)
return result.get("data") or []
async def generate_previews(
self,
ad_account_id: str,
src/adclaw/meta/insights.py
+172
View File
@@ -0,0 +1,172 @@
# -*- coding: utf-8 -*-
"""Turn Meta's insight rows into numbers that can be compared and stored.
Two things make raw rows unusable as they arrive:
1. **Every metric is a string.** Meta returns `"spend": "1234.56"`, not a
number. Comparing that to a guardrail limit with `>` would compare a
string to an int and raise, or worse, compare lexically.
2. **Leads are not a field.** They live inside the `actions` array as
`{"action_type": "lead", "value": "3"}`, alongside every other action
type. Cost per lead is likewise inside `cost_per_action_type`.
Spend is also in **major** currency units here (rupees, dollars), while
budgets and guardrails elsewhere in MaskanX are in **minor** units (paise,
cents) to match Graph's budget fields. `normalise_row` converts spend to
minor units so a guardrail limit and an observed spend are the same kind of
number. This is the single most important thing in this module: a factor of
100 in the wrong direction either never trips a guardrail or trips every
one of them.
These functions do no I/O.
"""
from __future__ import annotations
from typing import Any
# Meta names the same outcome differently depending on how the lead arrived.
# An on-Facebook lead form reports `lead`; a website conversion reports
# `offsite_conversion.fb_pixel_lead`. Both are leads for our purposes.
LEAD_ACTION_TYPES = (
"lead",
"onsite_conversion.lead_grouped",
"offsite_conversion.fb_pixel_lead",
"leadgen_grouped",
)
LINK_CLICK_ACTION_TYPES = ("link_click",)
PURCHASE_ACTION_TYPES = (
"purchase",
"offsite_conversion.fb_pixel_purchase",
"omni_purchase",
)
def _number(value: Any) -> float:
"""Parse one of Meta's stringified metrics. Anything unusable is 0.0."""
if value is None:
return 0.0
try:
return float(value)
except (TypeError, ValueError):
return 0.0
def _to_minor_units(major: float) -> int:
"""Convert a major-unit amount to minor units, rounding half **up**.
Not `round()`: Python rounds halves to even, so `round(0.5)` is 0. For
money that means an exact half-unit of spend can be recorded as nothing,
and under-reported spend is what lets a campaign slip past a stop-loss.
Adding 0.5 before truncating always rounds a tie upward instead.
Amounts here are spend and costs, which are never negative; the
`max(0.0, ...)` makes that explicit rather than silently mis-rounding
if Meta ever returns one.
"""
return int(max(0.0, major) * 100 + 0.5)
def _divide(spend: int, count: int) -> int | None:
"""Cost per unit, or None when there are no units yet.
None is not zero and not infinity: a campaign that has not converted
anyone yet has no cost per lead to speak of, and a guardrail must not
trip on it. Rounds half up, like every other money figure here.
"""
if not count:
return None
return int(spend / count + 0.5)
def action_value(rows: Any, action_types: tuple[str, ...]) -> float:
"""Sum the values of matching action rows.
Meta can report the same outcome under more than one action type in one
row, so this sums rather than taking the first match.
"""
if not isinstance(rows, list):
return 0.0
total = 0.0
for row in rows:
if not isinstance(row, dict):
continue
if row.get("action_type") in action_types:
total += _number(row.get("value"))
return total
def normalise_row(row: dict[str, Any]) -> dict[str, Any]:
"""Return one insight row as comparable numbers.
`spend` and every cost are in **minor** currency units, matching budgets
and guardrails. Derived costs are computed here rather than read from
Meta's `cost_per_action_type`, so that spend and cost per lead can never
disagree about the same window.
"""
spend_major = _number(row.get("spend"))
spend = _to_minor_units(spend_major)
impressions = int(_number(row.get("impressions")))
clicks = int(_number(row.get("clicks")))
actions = row.get("actions")
leads = int(action_value(actions, LEAD_ACTION_TYPES))
link_clicks = int(action_value(actions, LINK_CLICK_ACTION_TYPES))
purchases = int(action_value(actions, PURCHASE_ACTION_TYPES))
return {
"spend": spend,
"impressions": impressions,
"reach": int(_number(row.get("reach"))),
"frequency": _number(row.get("frequency")),
"clicks": clicks,
"link_clicks": link_clicks,
"ctr": _number(row.get("ctr")),
"cpc": _to_minor_units(_number(row.get("cpc"))),
"cpm": _to_minor_units(_number(row.get("cpm"))),
"leads": leads,
"purchases": purchases,
"cost_per_lead": _divide(spend, leads),
"cost_per_click": _divide(spend, clicks),
"date_start": row.get("date_start"),
"date_stop": row.get("date_stop"),
}
def total_metrics(rows: list[dict[str, Any]]) -> dict[str, Any]:
"""Aggregate several insight rows into one, as if they were one window.
Sums are summed; costs are recomputed from the totals rather than
averaged, since an average of per-day costs is not the cost over the
period.
An empty list gives a zeroed row rather than `{}`, so a caller can
compare against it without checking for emptiness first — a campaign
with no delivery has spent nothing, which is a fact, not a gap.
"""
totals = {
"spend": 0,
"impressions": 0,
"reach": 0,
"clicks": 0,
"link_clicks": 0,
"leads": 0,
"purchases": 0,
}
for row in rows:
normalised = normalise_row(row)
for key in totals:
totals[key] += normalised[key]
spend = totals["spend"]
totals["cost_per_lead"] = _divide(spend, totals["leads"])
totals["cost_per_click"] = _divide(spend, totals["clicks"])
totals["ctr"] = (
(totals["clicks"] / totals["impressions"] * 100)
if totals["impressions"]
else 0.0
)
return totals
tests/test_campaign_budget_approval.py
+152
View File
@@ -0,0 +1,152 @@
# -*- coding: utf-8 -*-
"""require_approval_for_budget_increase.
The hazard is narrow but real: an approver signs off on a daily budget, and
the number is raised before the campaign launches. The approval then names
a figure nobody agreed to.
"""
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from adclaw.app.routers import campaigns as campaigns_router
from adclaw.campaigns.models import CampaignSpec
class FakeRepo:
def __init__(self):
self.items: dict[str, CampaignSpec] = {}
self.events: list[dict] = []
async def get_campaign(self, campaign_id):
return self.items.get(campaign_id)
async def update_campaign(self, spec):
self.items[spec.id] = spec
return spec
async def add_event(self, campaign_id, event_type, actor=None,
reason=None, payload=None):
self.events.append({"campaign_id": campaign_id, "event_type": event_type})
class FakeMeta:
async def get_ad_account(self, ad_account_id):
return {"min_daily_budget": 1000, "currency": "INR"}
def _campaign(**overrides) -> CampaignSpec:
data = {
"id": "camp_1",
"name": "Q3 lead gen",
"status": "pending_approval",
"ad_account_id": "act_1",
"budget": {"daily_budget": 50000},
"guardrails": {"require_approval_for_budget_increase": True},
}
data.update(overrides)
return CampaignSpec(**data)
@pytest.fixture()
def env(monkeypatch):
repo = FakeRepo()
monkeypatch.setattr(campaigns_router, "get_repository", lambda: repo)
monkeypatch.setattr(campaigns_router, "get_meta_client", lambda: FakeMeta())
app = FastAPI()
app.include_router(campaigns_router.router, prefix="/api")
return TestClient(app), repo
def _put(client, **budget):
return client.put(
"/api/campaigns/camp_1",
json={"name": "Q3 lead gen", "budget": budget},
)
def test_raising_the_budget_while_awaiting_approval_is_refused(env):
client, repo = env
repo.items["camp_1"] = _campaign()
response = _put(client, daily_budget=90000)
assert response.status_code == 409
assert "awaiting approval" in response.json()["detail"]
assert repo.items["camp_1"].budget["daily_budget"] == 50000
def test_the_message_names_both_figures(env):
client, repo = env
repo.items["camp_1"] = _campaign()
detail = _put(client, daily_budget=90000).json()["detail"]
assert "50000" in detail and "90000" in detail
def test_lowering_the_budget_is_allowed(env):
"""A smaller number needs no re-approval."""
client, repo = env
repo.items["camp_1"] = _campaign()
response = _put(client, daily_budget=20000)
assert response.status_code == 200
assert repo.items["camp_1"].budget["daily_budget"] == 20000
def test_an_unchanged_budget_is_allowed(env):
client, repo = env
repo.items["camp_1"] = _campaign()
assert _put(client, daily_budget=50000).status_code == 200
def test_a_draft_has_no_pending_review_to_invalidate(env):
client, repo = env
repo.items["camp_1"] = _campaign(status="draft")
response = _put(client, daily_budget=90000)
assert response.status_code == 200
assert repo.items["camp_1"].budget["daily_budget"] == 90000
def test_without_the_flag_the_budget_can_be_raised(env):
client, repo = env
repo.items["camp_1"] = _campaign(guardrails={})
assert _put(client, daily_budget=90000).status_code == 200
def test_the_lifetime_budget_is_gated_too(env):
client, repo = env
repo.items["camp_1"] = _campaign(budget={"lifetime_budget": 500000})
response = _put(client, lifetime_budget=900000)
assert response.status_code == 409
assert "lifetime_budget" in response.json()["detail"]
def test_a_budget_field_absent_before_is_not_treated_as_an_increase():
"""Setting a budget field for the first time is not a raise.
Checked directly rather than through PUT: setting both a daily and a
lifetime budget is separately rejected by validation, which would mask
what this is testing.
"""
campaign = _campaign(budget={"daily_budget": 50000})
# No exception: lifetime_budget was not previously set.
campaigns_router._assert_budget_increase_allowed(
campaign, {"daily_budget": 50000, "lifetime_budget": 900000},
)
def test_dropping_a_budget_field_from_the_update_is_not_an_increase():
"""A partial update that omits the budget must not read as a change."""
campaign = _campaign(budget={"daily_budget": 50000})
campaigns_router._assert_budget_increase_allowed(campaign, {})
tests/test_campaign_enforce.py
+278
View File
@@ -0,0 +1,278 @@
# -*- coding: utf-8 -*-
"""The enforcement sweep: fetching spend, deciding, and acting on Meta.
The properties worth protecting are the failure modes, not the happy path:
Meta is paused before the local record changes, one campaign's failure does
not leave the others unguarded, and nothing that is not live is touched.
"""
import pytest
from adclaw.campaigns.enforce import (
DEFAULT_INTERVAL_SECONDS,
enforce_once,
guardrail_interval_seconds,
observed_spend,
)
from adclaw.campaigns.models import CampaignSpec
from adclaw.meta.client import MetaError
class FakeRepo:
def __init__(self, campaigns=()):
self.items = list(campaigns)
self.events: list[dict] = []
# Records the campaign status at the moment each event was written,
# so a test can prove Meta was called first.
self.saved_status: list[str] = []
async def list_campaigns(self, company_id=None, status=None):
if status is None:
return list(self.items)
return [c for c in self.items if c.status == status]
async def update_campaign_with_event(
self, spec, event_type, actor=None, reason=None, payload=None,
):
self.events.append({
"event_type": event_type,
"actor": actor,
"reason": reason,
"payload": payload,
})
self.saved_status.append(spec.status)
return spec
class FakeMeta:
def __init__(self, today=None, lifetime=None):
self.today = today if today is not None else []
self.lifetime = lifetime if lifetime is not None else []
self.status_calls: list[tuple[str, str]] = []
self.insight_calls: list[tuple[str, str]] = []
self.fail_insights_for: set[str] = set()
async def get_insights(self, object_id, *, date_preset=None, **kwargs):
if object_id in self.fail_insights_for:
raise MetaError("Rate limited", code=17)
self.insight_calls.append((object_id, date_preset))
return self.today if date_preset == "today" else self.lifetime
async def update_object_status(self, object_id, status):
self.status_calls.append((object_id, status))
def _campaign(**overrides) -> CampaignSpec:
data = {
"id": "camp_1",
"name": "Q3 lead gen",
"status": "live",
"ad_account_id": "act_1",
"meta_campaign_id": "meta_camp_1",
"budget": {"daily_budget": 50000},
"guardrails": {
"auto_pause_if_spend_reaches": 45000,
"stop_loss_enabled": True,
},
}
data.update(overrides)
return CampaignSpec(**data)
# --- interval ---
def test_interval_defaults_to_fifteen_minutes(monkeypatch):
monkeypatch.delenv("MASKANX_GUARDRAIL_INTERVAL_SECONDS", raising=False)
assert guardrail_interval_seconds() == DEFAULT_INTERVAL_SECONDS
def test_a_nonsense_interval_falls_back_to_the_default(monkeypatch):
"""A typo must not silently disable a safety system."""
monkeypatch.setenv("MASKANX_GUARDRAIL_INTERVAL_SECONDS", "every 5 mins")
assert guardrail_interval_seconds() == DEFAULT_INTERVAL_SECONDS
def test_zero_disables_enforcement(monkeypatch):
monkeypatch.setenv("MASKANX_GUARDRAIL_INTERVAL_SECONDS", "0")
assert guardrail_interval_seconds() == 0
# --- reading spend ---
async def test_observed_spend_reads_today_and_lifetime_separately():
meta = FakeMeta(
today=[{"spend": "460.00", "clicks": "20"}],
lifetime=[{"spend": "5000.00"}],
)
spend = await observed_spend(meta, _campaign())
assert spend["today_spend"] == 46000
assert spend["lifetime_spend"] == 500000
assert meta.insight_calls == [
("meta_camp_1", "today"),
("meta_camp_1", "maximum"),
]
async def test_a_campaign_with_no_delivery_reads_as_zero():
spend = await observed_spend(FakeMeta(), _campaign())
assert spend["today_spend"] == 0
assert spend["lifetime_spend"] == 0
assert spend["cost_per_lead"] is None
# --- the sweep ---
async def test_a_campaign_over_its_daily_limit_is_paused_on_meta():
repo = FakeRepo([_campaign()])
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
result = await enforce_once(repo, meta)
assert result.paused == ["camp_1"]
assert meta.status_calls == [("meta_camp_1", "PAUSED")]
assert repo.items[0].status == "paused"
async def test_the_event_names_the_rule_and_the_observed_value():
"""A campaign that stopped overnight has to be explainable."""
repo = FakeRepo([_campaign()])
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
await enforce_once(repo, meta)
event = repo.events[-1]
assert event["event_type"] == "campaign.guardrail_pause"
assert event["actor"] == "guardrail"
assert event["payload"]["rule"] == "auto_pause_if_spend_reaches"
assert event["payload"]["observed"] == 46000
assert event["payload"]["limit"] == 45000
async def test_meta_is_paused_before_the_local_record_is_written():
"""Recorded-as-paused-but-still-delivering is the outcome to avoid."""
repo = FakeRepo([_campaign()])
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
class OrderCheckingMeta(FakeMeta):
async def update_object_status(self, object_id, status):
# No event may have been written yet.
assert repo.events == []
await FakeMeta.update_object_status(self, object_id, status)
meta = OrderCheckingMeta(
today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}],
)
await enforce_once(repo, meta)
assert meta.status_calls == [("meta_camp_1", "PAUSED")]
assert repo.events
async def test_a_campaign_within_its_limits_is_left_alone():
repo = FakeRepo([_campaign()])
meta = FakeMeta(today=[{"spend": "100.00"}], lifetime=[{"spend": "100.00"}])
result = await enforce_once(repo, meta)
assert result.checked == 1
assert not result.acted
assert meta.status_calls == []
assert repo.events == []
@pytest.mark.parametrize(
"status", ["draft", "pending_approval", "approved", "synced", "paused", "stopped"],
)
async def test_only_live_campaigns_are_checked(status):
"""Nothing else can be spending, and acting would fight the operator."""
repo = FakeRepo([_campaign(status=status)])
meta = FakeMeta(today=[{"spend": "9999.00"}], lifetime=[{"spend": "9999.00"}])
result = await enforce_once(repo, meta)
assert result.checked == 0
assert meta.status_calls == []
async def test_a_live_campaign_without_a_meta_object_is_skipped():
repo = FakeRepo([_campaign(meta_campaign_id=None)])
meta = FakeMeta(today=[{"spend": "9999.00"}])
result = await enforce_once(repo, meta)
assert result.checked == 0
assert meta.status_calls == []
async def test_one_campaign_failing_does_not_leave_the_rest_unguarded():
"""A rate limit on the first must not skip the second."""
repo = FakeRepo([
_campaign(id="camp_1", meta_campaign_id="meta_1"),
_campaign(id="camp_2", meta_campaign_id="meta_2"),
])
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
meta.fail_insights_for = {"meta_1"}
result = await enforce_once(repo, meta)
assert result.failed == ["camp_1"]
assert result.paused == ["camp_2"]
assert ("meta_2", "PAUSED") in meta.status_calls
async def test_a_failed_pause_on_meta_leaves_the_campaign_live():
"""Better to keep showing it as live than to lose track of real spend."""
repo = FakeRepo([_campaign()])
class RefusingMeta(FakeMeta):
async def update_object_status(self, object_id, status):
raise MetaError("Permission denied", code=200)
meta = RefusingMeta(
today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}],
)
result = await enforce_once(repo, meta)
assert result.failed == ["camp_1"]
assert not result.acted
assert repo.items[0].status == "live"
assert repo.events == []
async def test_a_lifetime_breach_stops_the_campaign():
repo = FakeRepo([
_campaign(
guardrails={"max_campaign_spend": 500000, "stop_loss_enabled": True},
),
])
meta = FakeMeta(today=[{"spend": "10.00"}], lifetime=[{"spend": "6000.00"}])
result = await enforce_once(repo, meta)
assert result.stopped == ["camp_1"]
assert repo.items[0].status == "stopped"
assert repo.events[-1]["event_type"] == "campaign.guardrail_stop"
# Stopping still pauses on Meta: that is what actually halts delivery.
assert meta.status_calls == [("meta_camp_1", "PAUSED")]
async def test_the_trip_is_recorded_on_the_campaign_for_the_ui():
repo = FakeRepo([_campaign()])
meta = FakeMeta(today=[{"spend": "460.00"}], lifetime=[{"spend": "460.00"}])
await enforce_once(repo, meta)
trip = repo.items[0].advanced["last_guardrail_trip"]
assert trip["rule"] == "auto_pause_if_spend_reaches"
assert trip["observed"] == 46000
assert trip["at"]
tests/test_campaign_guardrails.py
+178
View File
@@ -0,0 +1,178 @@
# -*- coding: utf-8 -*-
"""Guardrail evaluation.
These rules are the last automatic defence against a campaign spending more
than intended, so the tests are mostly about the ways they could fail to
fire — or fire when they should not.
"""
import pytest
from adclaw.campaigns.guardrails import evaluate
def _rules(**overrides):
rules = {"stop_loss_enabled": True}
rules.update(overrides)
return rules
def test_no_rules_means_no_action():
assert evaluate({}, today_spend=999999, lifetime_spend=999999) is None
def test_daily_spend_at_the_limit_pauses():
"""'reaches' is inclusive: at the limit is at the limit."""
breach = evaluate(
_rules(auto_pause_if_spend_reaches=45000),
today_spend=45000,
lifetime_spend=45000,
)
assert breach is not None
assert breach.action == "pause"
assert breach.rule == "auto_pause_if_spend_reaches"
assert breach.observed == 45000
def test_daily_spend_below_the_limit_does_nothing():
assert (
evaluate(
_rules(auto_pause_if_spend_reaches=45000),
today_spend=44999,
lifetime_spend=44999,
)
is None
)
def test_lifetime_cap_stops_rather_than_pauses():
breach = evaluate(
_rules(max_campaign_spend=500000),
today_spend=1000,
lifetime_spend=500000,
)
assert breach.action == "stop"
assert breach.rule == "max_campaign_spend"
def test_lifetime_cap_needs_stop_loss_enabled():
"""The switch exists so the permanent rule can be turned off alone."""
assert (
evaluate(
_rules(max_campaign_spend=500000, stop_loss_enabled=False),
today_spend=1000,
lifetime_spend=999999,
)
is None
)
def test_disabling_stop_loss_leaves_the_pause_rules_working():
breach = evaluate(
_rules(
max_campaign_spend=500000,
auto_pause_if_spend_reaches=45000,
stop_loss_enabled=False,
),
today_spend=45000,
lifetime_spend=999999,
)
assert breach.action == "pause"
def test_the_most_severe_breach_wins():
"""Blowing both caps should stop the campaign, not merely pause it."""
breach = evaluate(
_rules(max_campaign_spend=500000, auto_pause_if_spend_reaches=1000),
today_spend=90000,
lifetime_spend=600000,
)
assert breach.action == "stop"
def test_cost_per_lead_over_the_limit_pauses():
breach = evaluate(
_rules(max_cost_per_lead=15000),
today_spend=60000,
lifetime_spend=60000,
cost_per_lead=20000,
)
assert breach.rule == "max_cost_per_lead"
assert breach.action == "pause"
def test_cost_rules_are_skipped_before_the_first_lead():
"""No leads yet is not an infinite cost per lead."""
assert (
evaluate(
_rules(max_cost_per_lead=100),
today_spend=60000,
lifetime_spend=60000,
cost_per_lead=None,
)
is None
)
def test_cost_per_lead_exactly_at_the_limit_is_allowed():
"""'max' is a ceiling that may be reached, unlike 'reaches'."""
assert (
evaluate(
_rules(max_cost_per_lead=15000),
today_spend=15000,
lifetime_spend=15000,
cost_per_lead=15000,
)
is None
)
def test_cost_per_click_over_the_limit_pauses():
breach = evaluate(
_rules(max_cost_per_click=2000),
today_spend=60000,
lifetime_spend=60000,
cost_per_click=2001,
)
assert breach.rule == "max_cost_per_click"
def test_a_limit_of_zero_is_honoured_not_read_as_unset():
"""Someone who caps spend at zero means it."""
breach = evaluate(
_rules(auto_pause_if_spend_reaches=0),
today_spend=0,
lifetime_spend=0,
)
assert breach is not None
assert breach.limit == 0
@pytest.mark.parametrize("value", [None, "not a number", True, False])
def test_an_unusable_limit_is_treated_as_absent(value):
assert (
evaluate(
_rules(auto_pause_if_spend_reaches=value),
today_spend=999999,
lifetime_spend=999999,
)
is None
)
def test_the_reason_names_the_rule_and_both_numbers():
breach = evaluate(
_rules(auto_pause_if_spend_reaches=45000),
today_spend=46000,
lifetime_spend=46000,
)
assert "auto_pause_if_spend_reaches" in breach.reason
assert "46000" in breach.reason
assert "45000" in breach.reason
tests/test_meta_insights.py
+120
View File
@@ -0,0 +1,120 @@
# -*- coding: utf-8 -*-
"""Normalising Meta's insight rows.
The unit conversion is the point of these tests. Meta reports spend in
major units ("12.34" rupees) while budgets and guardrails everywhere else
in MaskanX are in minor units (1234 paise). Getting that factor of 100
backwards either never trips a guardrail or trips all of them.
"""
import pytest
from adclaw.meta.insights import (
action_value,
normalise_row,
total_metrics,
)
def test_spend_is_converted_from_major_to_minor_units():
row = normalise_row({"spend": "12.34"})
assert row["spend"] == 1234
def test_spend_rounds_rather_than_truncating():
"""Truncating under-reports spend, which is how a stop-loss is missed."""
assert normalise_row({"spend": "10.999"})["spend"] == 1100
assert normalise_row({"spend": "0.005"})["spend"] == 1
def test_missing_and_unparseable_metrics_become_zero():
row = normalise_row({"spend": None, "impressions": "not a number"})
assert row["spend"] == 0
assert row["impressions"] == 0
def test_leads_are_dug_out_of_the_actions_array():
row = normalise_row({
"spend": "300.00",
"actions": [
{"action_type": "post_engagement", "value": "50"},
{"action_type": "lead", "value": "3"},
],
})
assert row["leads"] == 3
assert row["cost_per_lead"] == 10000 # 300.00 / 3 = 100.00
def test_leads_from_different_meta_action_types_are_summed():
"""An on-Facebook lead form and a pixel lead are both leads."""
row = normalise_row({
"actions": [
{"action_type": "lead", "value": "2"},
{"action_type": "offsite_conversion.fb_pixel_lead", "value": "5"},
],
})
assert row["leads"] == 7
def test_cost_per_lead_is_none_rather_than_infinite_with_no_leads():
"""A campaign that has not converted anyone yet must not trip a rule."""
row = normalise_row({"spend": "500.00", "actions": []})
assert row["leads"] == 0
assert row["cost_per_lead"] is None
def test_cost_per_click_is_none_with_no_clicks():
assert normalise_row({"spend": "500.00", "clicks": "0"})["cost_per_click"] is None
def test_action_value_ignores_malformed_rows():
assert action_value(None, ("lead",)) == 0.0
assert action_value(["not a dict"], ("lead",)) == 0.0
assert action_value([{"action_type": "lead"}], ("lead",)) == 0.0
# --- aggregation ---
def test_totals_of_no_rows_are_zero_not_empty():
"""No delivery means nothing spent, which is a fact, not a gap."""
totals = total_metrics([])
assert totals["spend"] == 0
assert totals["leads"] == 0
assert totals["cost_per_lead"] is None
def test_totals_sum_across_rows():
totals = total_metrics([
{"spend": "10.00", "clicks": "5", "actions": [{"action_type": "lead", "value": "1"}]},
{"spend": "30.00", "clicks": "15", "actions": [{"action_type": "lead", "value": "3"}]},
])
assert totals["spend"] == 4000
assert totals["clicks"] == 20
assert totals["leads"] == 4
def test_cost_per_lead_is_recomputed_from_totals_not_averaged():
"""An average of daily costs is not the cost over the period."""
totals = total_metrics([
# Day one: 1 lead at 100.00
{"spend": "100.00", "actions": [{"action_type": "lead", "value": "1"}]},
# Day two: 9 leads at 100.00 total, so ~11.11 each
{"spend": "100.00", "actions": [{"action_type": "lead", "value": "9"}]},
])
# 200.00 over 10 leads is 20.00, not the 55.55 an average would give.
assert totals["cost_per_lead"] == 2000
@pytest.mark.parametrize("impressions,expected", [("0", 0.0), ("1000", 5.0)])
def test_ctr_is_recomputed_and_safe_at_zero_impressions(impressions, expected):
totals = total_metrics([{"clicks": "50", "impressions": impressions}])
assert totals["ctr"] == expected