"""Optimization Engine — evaluates and executes optimization rules on ad campaigns.
Runs scheduled checks on campaigns, ad groups, ads, and keywords based on
configurable rules. Every action is logged for audit and rollback.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, timezone, timedelta
from typing import Any, Dict, List, Optional
from ...models import db, AdCampaign, AdMetric, OptimizationRule, OptimizationLog
logger = logging.getLogger(__name__)
class OptimizationEngine:
"""Evaluate and execute optimization rules.
Supports rule types:
- kill_high_cpa: Pause campaigns where CPA > target * multiplier
- pause_zero_conv: Pause campaigns with 0 conversions after min_spend
- scale_low_cpa: Increase budget on campaigns performing below target CPA
- kill_low_ctr: Pause ads/keywords with CTR below threshold
- budget_reallocation: Shift budget from poor to good performers
"""
# Hard safety limits that cannot be overridden by rules
MAX_BUDGET_INCREASE_PCT = 100 # Never increase budget more than 100%
MIN_CAMPAIGNS_ACTIVE = 1 # Never kill the last active campaign
MIN_SPEND_THRESHOLD = 50 # Minimum spend before making decisions (avoid noise)
def __init__(self, company_id: str):
self.company_id = company_id
def run_rule(self, rule: OptimizationRule) -> Dict[str, Any]:
"""Execute a single optimization rule against all relevant campaigns.
Args:
rule: OptimizationRule instance
Returns:
Dict with summary of actions taken
"""
start = time.monotonic()
params = rule.params_json or {}
actions = []
try:
handler = getattr(self, f'_handle_{rule.rule_type}', None)
if not handler:
return {
'error': f'Unknown rule type: {rule.rule_type}',
'supported': [
'kill_high_cpa', 'pause_zero_conv', 'scale_low_cpa',
'kill_low_ctr', 'budget_reallocation',
],
}
actions = handler(rule, params)
except Exception as e:
logger.error("Rule %s failed for company %s: %s", rule.id, self.company_id, e, exc_info=True)
return {
'error': str(e),
'actions': actions,
}
# Update rule stats
rule.last_run_at = datetime.now(timezone.utc)
rule.total_actions += len(actions)
db.session.commit()
duration_ms = int((time.monotonic() - start) * 1000)
return {
'rule_id': rule.id,
'rule_name': rule.name,
'rule_type': rule.rule_type,
'actions_taken': len(actions),
'actions': actions,
'duration_ms': duration_ms,
}
def run_all_active(self) -> Dict[str, Any]:
"""Run all active optimization rules for this company."""
rules = OptimizationRule.query.filter_by(
company_id=self.company_id,
is_active=True,
).all()
results = []
total_actions = 0
for rule in rules:
result = self.run_rule(rule)
results.append(result)
total_actions += result.get('actions_taken', 0)
return {
'rules_run': len(rules),
'total_actions': total_actions,
'results': results,
}
def _handle_kill_high_cpa(
self, rule: OptimizationRule, params: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""Pause campaigns where CPA exceeds target * multiplier.
Params:
cpa_multiplier: Pause if CPA > target_cpa * multiplier (default: 2.0)
lookback_days: Analyze last N days of data (default: 14)
min_spend: Minimum spend to consider (default: 100)
action: 'pause' or 'delete' (default: 'pause')
"""
multiplier = params.get('cpa_multiplier', 2.0)
lookback_days = params.get('lookback_days', 14)
min_spend = params.get('min_spend', 100)
action = params.get('action', 'pause')
since = datetime.now(timezone.utc) - timedelta(days=lookback_days)
campaigns = AdCampaign.query.filter_by(
company_id=self.company_id,
status='ENABLED',
).all()
actions = []
# Count active campaigns for safety
active_count = len(campaigns)
for campaign in campaigns:
# Get metrics for this campaign in the lookback window
metrics = AdMetric.query.filter_by(
company_id=self.company_id,
external_campaign_id=campaign.external_id,
).filter(
AdMetric.metric_date >= since.date()
).all()
if not metrics:
continue
total_spend = sum(m.spend for m in metrics)
total_conversions = sum(m.conversions for m in metrics)
# Skip if not enough data
if total_spend < min_spend or total_conversions == 0:
continue
cpa = total_spend / total_conversions
# Get target CPA from campaign metadata or template
target_cpa = campaign.metadata_json.get('target_cpa')
if not target_cpa:
continue
threshold = target_cpa * multiplier
if cpa > threshold:
# Safety: don't kill the last campaign
if active_count <= self.MIN_CAMPAIGNS_ACTIVE:
logger.warning(
"Skipping kill rule: only %d active campaign(s) remaining",
active_count,
)
continue
# Execute the action
success, error_msg = self._execute_action(
rule, campaign, action,
reason=f"CPA ${cpa:.2f} > ${threshold:.2f} target × {multiplier}x over {lookback_days}d, spend ${total_spend:.2f}",
before={'status': 'ENABLED', 'budget': campaign.budget, 'cpa': round(cpa, 2)},
)
if success:
actions.append({
'action': action,
'campaign': campaign.name,
'cpa': round(cpa, 2),
'threshold': round(threshold, 2),
'target_cpa': target_cpa,
})
if action == 'pause':
active_count -= 1
return actions
def _handle_pause_zero_conv(
self, rule: OptimizationRule, params: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""Pause campaigns with 0 conversions after minimum spend threshold.
Params:
min_spend: Minimum spend before pausing (default: 100)
lookback_days: Analyze last N days (default: 14)
"""
min_spend = params.get('min_spend', 100)
lookback_days = params.get('lookback_days', 14)
since = datetime.now(timezone.utc) - timedelta(days=lookback_days)
campaigns = AdCampaign.query.filter_by(
company_id=self.company_id,
status='ENABLED',
).all()
active_count = len(campaigns)
actions = []
for campaign in campaigns:
metrics = AdMetric.query.filter_by(
company_id=self.company_id,
external_campaign_id=campaign.external_id,
).filter(
AdMetric.metric_date >= since.date()
).all()
if not metrics:
continue
total_spend = sum(m.spend for m in metrics)
total_conversions = sum(m.conversions for m in metrics)
if total_spend >= min_spend and total_conversions == 0:
if active_count <= self.MIN_CAMPAIGNS_ACTIVE:
continue
success, _ = self._execute_action(
rule, campaign, 'pause',
reason=f"Zero conversions over {lookback_days}d with ${total_spend:.2f} spend",
before={'status': 'ENABLED', 'spend': round(total_spend, 2), 'conversions': 0},
)
if success:
actions.append({
'action': 'pause',
'campaign': campaign.name,
'spend': round(total_spend, 2),
'conversions': 0,
})
active_count -= 1
return actions
def _handle_scale_low_cpa(
self, rule: OptimizationRule, params: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""Increase budget on campaigns performing below target CPA.
Params:
cpa_multiplier: Scale if CPA < target_cpa * multiplier (default: 0.7)
lookback_days: Analyze last N days (default: 14)
min_spend: Minimum spend to consider (default: 200)
max_budget_increase_pct: Cap budget increase (default: 50, max: 100)
action: 'increase_budget' (default)
"""
multiplier = params.get('cpa_multiplier', 0.7)
lookback_days = params.get('lookback_days', 14)
min_spend = params.get('min_spend', 200)
max_increase_pct = min(params.get('max_budget_increase_pct', 50), self.MAX_BUDGET_INCREASE_PCT)
since = datetime.now(timezone.utc) - timedelta(days=lookback_days)
campaigns = AdCampaign.query.filter_by(
company_id=self.company_id,
status='ENABLED',
).all()
actions = []
for campaign in campaigns:
metrics = AdMetric.query.filter_by(
company_id=self.company_id,
external_campaign_id=campaign.external_id,
).filter(
AdMetric.metric_date >= since.date()
).all()
if not metrics:
continue
total_spend = sum(m.spend for m in metrics)
total_conversions = sum(m.conversions for m in metrics)
if total_spend < min_spend or total_conversions == 0:
continue
cpa = total_spend / total_conversions
target_cpa = campaign.metadata_json.get('target_cpa')
if not target_cpa:
continue
threshold = target_cpa * multiplier
if cpa < threshold:
current_budget = campaign.budget or 0
new_budget = current_budget * (1 + max_increase_pct / 100)
success, _ = self._execute_action(
rule, campaign, 'scaled',
reason=f"CPA ${cpa:.2f} < ${threshold:.2f} (target ${target_cpa:.2f} × {multiplier}), "
f"increasing budget ${current_budget:.2f} → ${new_budget:.2f}",
before={'budget': current_budget, 'cpa': round(cpa, 2)},
after={'budget': new_budget},
)
if success:
campaign.budget = new_budget
actions.append({
'action': 'scaled',
'campaign': campaign.name,
'old_budget': current_budget,
'new_budget': round(new_budget, 2),
'cpa': round(cpa, 2),
})
return actions
def _handle_kill_low_ctr(
self, rule: OptimizationRule, params: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""Pause ads/keywords with CTR below threshold.
Params:
min_ctr: Minimum CTR percentage (default: 1.0)
lookback_days: Analyze last N days (default: 14)
min_impressions: Minimum impressions to consider (default: 100)
target_level: 'ad' or 'keyword' (default: 'keyword')
"""
from ...models import AdKeyword, AdCreative, Connector
min_ctr = params.get('min_ctr', 1.0)
lookback_days = params.get('lookback_days', 14)
min_impressions = params.get('min_impressions', 100)
target_level = params.get('target_level', rule.target_level or 'keyword')
# SMS compliance keywords — never optimize these
from ...utils.sms_keywords import KNOWN_KEYWORDS as SMS_KEYWORDS
sms_keywords = {k.lower() for k in SMS_KEYWORDS}
logger.info(
"Low CTR rule: company=%s, min_ctr=%.2f%%, target=%s, lookback=%dd",
self.company_id, min_ctr, target_level, lookback_days,
)
actions: List[Dict[str, Any]] = []
if target_level == 'keyword':
actions = self._kill_low_ctr_keywords(
rule, min_ctr, lookback_days, min_impressions, sms_keywords,
)
elif target_level == 'ad':
actions = self._kill_low_ctr_ads(
rule, min_ctr, lookback_days, min_impressions, sms_keywords,
)
else:
logger.warning("Unknown target_level for kill_low_ctr: %s — skipping", target_level)
return actions
def _kill_low_ctr_keywords(
self,
rule: OptimizationRule,
min_ctr: float,
lookback_days: int,
min_impressions: int,
sms_keywords: set,
) -> List[Dict[str, Any]]:
"""Evaluate and pause keywords with CTR below threshold.
Uses AdKeyword model fields: text, match_type, status,
total_impressions, total_clicks, total_spend, avg_ctr,
external_campaign_id, external_ad_group_id.
"""
from ...models import AdKeyword, AdCampaign, Connector
# Find connected Google Ads connector
connector = Connector.query.filter_by(
company_id=self.company_id,
service='google_ads',
status='connected',
).first()
if not connector:
logger.info("No connected Google Ads connector for keyword CTR check")
return []
# Get all keywords for this company's campaigns
keywords = AdKeyword.query.filter_by(company_id=self.company_id).all()
actions: List[Dict[str, Any]] = []
for kw in keywords:
# Skip SMS compliance keywords
kw_text = (kw.text or '').lower().strip()
if kw_text in sms_keywords:
continue
# Skip already-paused keywords
if kw.status == 'paused':
continue
impressions = kw.total_impressions or 0
clicks = kw.total_clicks or 0
# Skip if not enough impressions for statistical significance
if impressions < min_impressions:
continue
ctr = (clicks / impressions * 100) if impressions > 0 else 0
if ctr < min_ctr:
# Find the parent campaign for connector access
campaign = AdCampaign.query.filter_by(
company_id=self.company_id,
external_id=kw.external_campaign_id,
).first()
if not campaign:
logger.warning("Keyword '%s' has no parent campaign", kw.text)
continue
# Execute the pause
success, error_msg = self._execute_keyword_action(
rule,
kw,
campaign,
'pause',
connector,
reason=f"CTR {ctr:.2f}% < {min_ctr:.2f}% threshold over {lookback_days}d "
f"({impressions} impressions, {clicks} clicks)",
before={
'status': kw.status,
'ctr': round(ctr, 4),
'impressions': impressions,
'clicks': clicks,
'spend': round(kw.total_spend or 0, 2),
},
)
if success:
actions.append({
'action': 'pause',
'target_type': 'keyword',
'keyword': kw.text,
'match_type': kw.match_type,
'ctr': round(ctr, 4),
'min_ctr': min_ctr,
'impressions': impressions,
'clicks': clicks,
'campaign': campaign.name,
})
return actions
def _kill_low_ctr_ads(
self,
rule: OptimizationRule,
min_ctr: float,
lookback_days: int,
min_impressions: int,
sms_keywords: set,
) -> List[Dict[str, Any]]:
"""Evaluate and pause ad creatives with CTR below threshold."""
from ...models import AdCreative, AdCampaign, AdMetric
ads = AdCreative.query.filter_by(
company_id=self.company_id,
is_active=True,
).all()
since = datetime.now(timezone.utc) - timedelta(days=lookback_days)
actions: List[Dict[str, Any]] = []
for ad in ads:
impressions = ad.total_impressions or 0
clicks = ad.total_clicks or 0
if impressions < min_impressions:
continue
ctr = ad.avg_ctr or 0
if ctr < min_ctr:
campaign = AdCampaign.query.filter_by(
company_id=self.company_id,
id=ad.campaign_id,
).first()
if not campaign:
continue
success, _ = self._execute_action(
rule,
campaign,
'pause',
reason=f"Ad CTR {ctr:.2f}% < {min_ctr:.2f}% threshold "
f"({impressions} impressions, {clicks} clicks)",
before={
'ad_id': ad.id,
'status': 'active',
'ctr': round(ctr, 4),
'impressions': impressions,
},
)
if success:
actions.append({
'action': 'pause',
'target_type': 'ad',
'ad_id': ad.id,
'headlines': ad.headlines[:1] if ad.headlines else [],
'ctr': round(ctr, 4),
'min_ctr': min_ctr,
})
return actions
def _execute_keyword_action(
self,
rule: OptimizationRule,
keyword: "AdKeyword",
campaign: "AdCampaign",
action: str,
connector,
reason: str,
before: Optional[Dict[str, Any]] = None,
after: Optional[Dict[str, Any]] = None,
) -> tuple[bool, str]:
"""Execute an action on a keyword via the ad platform connector.
Calls the appropriate connector method (pause_keyword, enable_keyword)
and logs the result.
Returns:
(success, error_message) tuple
"""
from ...models import OptimizationLog
log_status = 'success'
log_error = ''
connector_error = ''
try:
if action == 'pause':
result = connector.pause_keyword(
ad_group_resource_name=keyword.external_ad_group_id or campaign.external_id,
keyword=keyword.text or '',
match_type=(keyword.match_type or 'broad').upper(),
)
if result.get('status') == 'error':
connector_error = result.get('error', 'pause_keyword failed')
if not connector_error:
# Update local state
keyword.status = 'paused'
after = after or {'status': 'paused'}
elif action == 'enable':
result = connector.enable_keyword(
ad_group_resource_name=keyword.external_ad_group_id or campaign.external_id,
keyword=keyword.text or '',
match_type=(keyword.match_type or 'broad').upper(),
)
if result.get('status') == 'error':
connector_error = result.get('error', 'enable_keyword failed')
if not connector_error:
keyword.status = 'ENABLED'
after = after or {'status': 'ENABLED'}
else:
connector_error = f"Unsupported keyword action: {action}"
except Exception as e:
log_error = f"Keyword {action} failed: {e}"
logger.error(log_error, exc_info=True)
if connector_error:
log_status = 'error'
log_error = connector_error
# Log the action
log = OptimizationLog(
company_id=self.company_id,
rule_id=rule.id,
source_service=campaign.source_service,
target_type='keyword',
target_id=keyword.id,
target_name=keyword.text or '',
action=action,
before_json=before or {},
after_json=after or {},
reason=reason,
status=log_status,
error_message=log_error,
)
db.session.add(log)
db.session.flush()
logger.info(
"Keyword optimization: %s '%s' (%s) — %s",
action, keyword.text or '', keyword.id, reason,
)
return not log_error, log_error
def _handle_budget_reallocation(
self, rule: OptimizationRule, params: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""Shift budget from poor performers to good performers.
Params:
lookback_days: Analyze last N days (default: 14)
min_spend: Minimum spend threshold (default: 100)
max_shift_pct: Max percentage to shift from any single campaign (default: 30)
top_percentile: Top N% performers receive reallocated budget (default: 33)
"""
lookback_days = params.get('lookback_days', 14)
min_spend = params.get('min_spend', 100)
max_shift_pct = min(params.get('max_shift_pct', 30), 50)
since = datetime.now(timezone.utc) - timedelta(days=lookback_days)
campaigns = AdCampaign.query.filter_by(
company_id=self.company_id,
status='ENABLED',
).all()
if len(campaigns) < 2:
return []
# Calculate CPA for each campaign
campaign_cpa = []
for campaign in campaigns:
metrics = AdMetric.query.filter_by(
company_id=self.company_id,
external_campaign_id=campaign.external_id,
).filter(
AdMetric.metric_date >= since.date()
).all()
if not metrics:
continue
total_spend = sum(m.spend for m in metrics)
total_conversions = sum(m.conversions for m in metrics)
if total_spend < min_spend or total_conversions == 0:
continue
cpa = total_spend / total_conversions
campaign_cpa.append((campaign, cpa, total_spend))
if len(campaign_cpa) < 2:
return []
# Sort by CPA (lower is better)
campaign_cpa.sort(key=lambda x: x[1])
# Determine threshold for good vs poor performers
n = len(campaign_cpa)
good_count = max(1, n // 3) # Top third
good_campaigns = campaign_cpa[:good_count]
poor_campaigns = campaign_cpa[good_count:]
# Calculate budget to reallocate from poor performers
total_reallocation = 0
actions = []
for campaign, cpa, spend in poor_campaigns:
shift_amount = (campaign.budget or 0) * (max_shift_pct / 100)
if shift_amount > 0:
total_reallocation += shift_amount
actions.append({
'action': 'rebudgeted',
'campaign': campaign.name,
'shift_amount': round(shift_amount, 2),
'direction': 'decrease',
})
# Distribute to good performers
if total_reallocation > 0 and good_campaigns:
share = total_reallocation / len(good_campaigns)
for campaign, cpa, spend in good_campaigns:
actions.append({
'action': 'rebudgeted',
'campaign': campaign.name,
'shift_amount': round(share, 2),
'direction': 'increase',
})
return actions
def _execute_action(
self,
rule: OptimizationRule,
campaign: AdCampaign,
action: str,
reason: str,
before: Optional[Dict[str, Any]] = None,
after: Optional[Dict[str, Any]] = None,
) -> tuple[bool, str]:
"""Execute an action on a campaign and log it.
Calls the appropriate ad platform connector (Google Ads, Facebook Ads)
to perform the action on the live platform, then records the result.
Returns:
(success, error_message) tuple
"""
log_status = 'success'
log_error = ''
# Attempt to call the connector API for platform-side action
connector_error = ''
if campaign.source_service in ('google_ads', 'facebook_ads'):
connector_error = self._call_connector_action(
campaign, action,
)
# Update local campaign status regardless of connector result
if action == 'pause':
campaign.status = 'PAUSED'
elif action == 'delete':
campaign.status = 'REMOVED'
# For 'scaled', the DB budget is already updated by the caller before this
if connector_error:
log_status = 'error'
log_error = connector_error
# Log the action
log = OptimizationLog(
company_id=self.company_id,
rule_id=rule.id,
source_service=campaign.source_service,
target_type='campaign',
target_id=campaign.external_id,
target_name=campaign.name,
action=action,
before_json=before or {},
after_json=after or {},
reason=reason,
status=log_status,
error_message=log_error,
)
db.session.add(log)
db.session.flush()
logger.info(
"Optimization action: %s on %s (%s) — %s",
action, campaign.name, campaign.external_id, reason,
)
return not connector_error, connector_error
def _call_connector_action(
self,
campaign: AdCampaign,
action: str,
) -> str:
"""Call the ad platform connector to perform an action.
Returns:
Empty string on success, error message string on failure.
"""
from app.models import Connector
# Find the active connector for this service/company
connector = Connector.query.filter_by(
company_id=self.company_id,
service=campaign.source_service,
status='connected',
).first()
if not connector:
msg = f"No connected {campaign.source_service} connector for company {self.company_id[:8]}"
logger.warning("Optimization skipped %s: %s", campaign.name, msg)
return msg
try:
from app.connectors import build_connector
conn = build_connector(
service=campaign.source_service,
company_id=self.company_id,
config=connector.config,
connector_id=connector.id,
)
except Exception as e:
msg = f"Failed to build connector: {e}"
logger.error(msg)
return msg
try:
if action == 'pause':
result = conn.pause_campaign(campaign.external_id)
if not result.get('success') and result.get('status') != 'success':
return result.get('error', 'pause_campaign returned failure')
elif action == 'delete':
# Most platforms don't support hard-delete on campaigns
# Fall back to pause
result = conn.pause_campaign(campaign.external_id)
if not result.get('success') and result.get('status') != 'success':
return result.get('error', 'pause_campaign returned failure')
elif action == 'scaled':
# Budget update
new_budget = campaign.budget or 0
result = conn.update_campaign_budget(
campaign.external_id,
new_budget,
)
if not result.get('success') and result.get('status') != 'success':
return result.get('error', 'update_campaign_budget returned failure')
else:
return f"Unsupported action: {action}"
return '' # success
except Exception as e:
msg = f"Connector action {action} failed for {campaign.name}: {e}"
logger.error(msg, exc_info=True)
return msg