Search term optimizer logic simplified for direct llm analysis

app/ads1/search_term_optimizer.py

@@ -1,99 +1,110 @@
-import pandas as pd
-from app.recs.generate import TARGET_CPL, generate_explanation, is_bad_llm_output
 import json
+import pandas as pd
+from app.recs.generate import generate_explanation, is_bad_llm_output
 
+# -------------------------
+# Feature engineering only
+# -------------------------
 def build_search_term_features(df: pd.DataFrame) -> pd.DataFrame:
     df = df.copy()
 
     df["cost"] = df["cost"].fillna(0)
     df["clicks"] = df["clicks"].fillna(0)
     df["impressions"] = df["impressions"].fillna(0)
+
     if "conversions" not in df.columns:
         df["conversions"] = 0
 
     df["conversions"] = df["conversions"].fillna(0)
-    df["ctr"] = ((df["clicks"] / df["impressions"])*100).replace(0, 1)
-    df["cvr"] = ((df["conversions"] / df["clicks"])*100).replace(0, 1)
+
+    # Core metrics
+    df["ctr"] = (df["clicks"] / df["impressions"].replace(0, 1)) * 100
+    df["cvr"] = (df["conversions"] / df["clicks"].replace(0, 1)) * 100
     df["cpc"] = df["cost"] / df["clicks"].replace(0, 1)
     df["cpa"] = df["cost"] / df["conversions"].replace(0, 1)
 
     return df
 
-def _search_terms_for_campaign(dfs: dict, campaign_name: str | None) -> pd.DataFrame:
+
+# -------------------------
+# Optional: lightweight pruning (NOT rule-based logic)
+# -------------------------
+def prepare_context_df(df: pd.DataFrame, max_rows: int = 200) -> pd.DataFrame:
+    """
+    Instead of semantic filtering, we just cap size for token control.
+    Keeps high-variance distribution intact.
+    """
+    return df.sort_values("cost", ascending=False).head(max_rows)
+
+# -------------------------
+# Prompt (LLM owns all reasoning now)
+# -------------------------
+def build_search_optimizer_prompt(context: dict) -> str:
+    payload = json.dumps(context, indent=2, default=str)
+    name = context.get("campaign_name", "this campaign")
+
+    return (
+        f"You are an expert Google Ads search term optimization strategist.\n\n"
+        f"Analyze search term performance for {name}.\n\n"
+        "Your job is to provide 3 to 5 actionable insights.\n"
+        "Focus on:\n"
+        "- wasted spend search terms\n"
+        "- high intent converting terms\n"
+        "- scaling opportunities\n"
+        "- terms to pause or reduce bids\n"
+        "- unusual CTR / CPA / conversion patterns\n\n"
+        "Rules:\n"
+        "- Be specific and actionable\n"
+        "- Use simple language\n"
+        "- One insight per bullet point\n"
+        "- Start each line with '- '\n"
+        "- No intro sentence, no summary, no JSON\n\n"
+        f"Data:\n{payload}"
+    )
+
+# -------------------------
+# Context builder (FULL DATA approach)
+# -------------------------
+def build_search_optimizer_context(dfs: dict, campaign_name: str | None = None):
     df = dfs["search_terms"].copy()
-    if (campaign_name
-        and "campaign_name" in df.columns):
-            df = df[df["campaign_name"] == campaign_name]
-    return df
 
-def detect_negative_terms(df):
-    return df[
-        (df["clicks"] >= 20) &
-        (df["cost"] >= 5) &
-        (df["conversions"] == 0)
-    ]
+    if campaign_name and "campaign_name" in df.columns:
+        df = df[df["campaign_name"] == campaign_name]
 
-def detect_review_terms(df):
-    return df[
-        (df["clicks"] >= 10) &
-        (df["ctr"] < 2) &
-        (df["conversions"] == 0)
-    ]
+    df = build_search_term_features(df)
+    df = prepare_context_df(df)
 
-def detect_scaling_terms(df):
-    return df[(df["conversions"] > 0)].sort_values("cpa", ascending=True)
+    return {
+        "campaign_name": campaign_name,
+        "search_terms": df.to_dict("records")  # FULL dataset context (bounded)
+    }
 
-def build_search_optimizer_prompt(context: dict) -> str:
-    # Convert data to clean JSON string strings for better LLM readability
-    neg_json = json.dumps(context['negative_keywords'], indent=2)
-    rev_json = json.dumps(context['review_terms'], indent=2)
-    win_json = json.dumps(context['winning_terms'], indent=2)
 
-    return f"""You are an expert Google Ads optimization engine.
-        Analyze the following three categories of search terms for the campaign "{context['campaign_name'] or 'All Campaigns'}":
+# -------------------------
+# Runner
+# -------------------------
+def run_search_term_optimizer(dfs: dict, campaign_name: str | None = None) -> str:
+    print("\n🚀 [search_term_optimizer] STARTED", flush=True)
 
-        ### HIGH WASTE TERMS (Add as negatives)
-        {neg_json}
+    if not dfs or "search_terms" not in dfs:
+        return "⚠️ No search term data — select a campaign first."
 
-        ### LOW CTR REVIEW TERMS (Needs attention)
-        {rev_json}
+    context = build_search_optimizer_context(dfs, campaign_name)
 
-        ### WINNING SCALING TERMS (Profitable)
-        {win_json}
+    print("🧠 [search_term_optimizer] context built", flush=True)
 
-        Provide your analysis in a strict JSON format with this exact schema:
-        {{
-            "waste_reduction_actions": ["action 1", "action 2"],
-            "scaling_opportunities": ["opportunity 1", "opportunity 2"],
-            "immediate_negative_keywords": ["keyword 1", "keyword 2"]
-        }}
-        Return ONLY valid JSON. Do not include markdown code blocks, explanations, or extra text."""
+    prompt = build_search_optimizer_prompt(context)
 
-def build_search_optimizer_context(dfs: dict, campaign_name: str | None = None):
-    df = _search_terms_for_campaign(dfs, campaign_name)
-    df = build_search_term_features(df)
-    negatives = detect_negative_terms(df)
-    review = detect_review_terms(df)
-    winners = detect_scaling_terms(df)
+    print("✍️ [search_term_optimizer] prompt built", flush=True)
 
-    return {
-        "campaign_name": campaign_name,
-        "negative_keywords": negatives.head(10).to_dict("records"),
-        "review_terms": review.head(10).to_dict("records"),
-        "winning_terms": winners.head(10).to_dict("records"),
-    }
+    result = generate_explanation(prompt)
 
-def run_search_term_optimizer(dfs: dict,campaign_name: str | None = None) -> str:
-    print("\n🚀 [search_term_card] STARTED", flush=True)
-    if not dfs:
+    if is_bad_llm_output(result):
+        print("⚠️ [search_term_optimizer] LLM fallback triggered", flush=True)
         return (
-            "⚠️ No campaign data — "
-            "select a campaign first."
+            "- Unable to generate insights right now.\n"
+            "- Try again or check data quality."
         )
-    context = build_search_optimizer_context(dfs,campaign_name)
-    print("🧠 [search_term_card] context built",flush=True)
-    prompt = build_search_optimizer_prompt(context)
-    print("✍️ [search_term_card] prompt built",flush=True)
-    result = generate_explanation(prompt)
-    print("📤 [search_term_card] result received", flush=True)
+
+    print("📤 [search_term_optimizer] result received", flush=True)
     return result