Databricks Unity Catalog Data Pipeline ETL — สร้าง Medallion Architecture

Andrew Perpetua และ Facebook Ads

Andrew Perpetua เป็นผู้เชี่ยวชาญ Facebook Ads เคยทำงานที่ Meta ดูแล Ads Products มีประสบการณ์มากกว่า 10 ปี Performance Marketing แชร์ความรู้ผ่าน Blog และ Social Media

แนวคิดหลักของ Andrew Perpetua คือใช้ Data-driven Approach ทดสอบ Creative อย่างเป็นระบบ ให้ Algorithm ทำงาน ไม่ Over-optimize ด้วยมือ

Campaign Structure และ Setup

# === Facebook Ads Campaign Structure ===
# ใช้ Facebook Marketing API (Python SDK)
# pip install facebook-business

# campaign_structure.py
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum

class CampaignObjective(Enum):
    CONVERSIONS = "OUTCOME_SALES"
    LEADS = "OUTCOME_LEADS"
    TRAFFIC = "OUTCOME_TRAFFIC"
    AWARENESS = "OUTCOME_AWARENESS"
    ENGAGEMENT = "OUTCOME_ENGAGEMENT"

class BidStrategy(Enum):
    LOWEST_COST = "LOWEST_COST_WITHOUT_CAP"
    COST_CAP = "COST_CAP"
    BID_CAP = "BID_CAP"
    ROAS_TARGET = "MINIMUM_ROAS"

@dataclass
class AdCreative:
    name: str
    headline: str
    description: str
    cta: str  # SHOP_NOW, LEARN_MORE, SIGN_UP
    image_url: Optional[str] = None
    video_url: Optional[str] = None
    format: str = "single_image"  # single_image, video, carousel

@dataclass
class AdSetConfig:
    name: str
    targeting: Dict
    budget_daily: float
    optimization_goal: str  # CONVERSIONS, LANDING_PAGE_VIEWS, LINK_CLICKS
    bid_strategy: BidStrategy = BidStrategy.LOWEST_COST
    creatives: List[AdCreative] = field(default_factory=list)

@dataclass
class CampaignConfig:
    name: str
    objective: CampaignObjective
    budget_daily: float
    use_cbo: bool = True  # Campaign Budget Optimization
    ad_sets: List[AdSetConfig] = field(default_factory=list)

class FacebookAdsManager:
    """Facebook Ads Campaign Manager"""

    def __init__(self, account_id, access_token):
        self.account_id = account_id
        self.token = access_token

    def create_campaign(self, config: CampaignConfig):
        """สร้าง Campaign"""
        # Facebook Marketing API call
        # campaign = AdAccount(account_id).create_campaign(params={
        #     'name': config.name,
        #     'objective': config.objective.value,
        #     'special_ad_categories': [],
        #     'campaign_budget_optimization': config.use_cbo,
        #     'daily_budget': int(config.budget_daily * 100),
        #     'bid_strategy': 'LOWEST_COST_WITHOUT_CAP',
        # })
        print(f"Campaign: {config.name}")
        print(f"  Objective: {config.objective.value}")
        print(f"  Budget: /day")
        print(f"  CBO: {config.use_cbo}")
        return config

    def create_targeting(self, countries, age_min=18, age_max=65,
                        genders=None, interests=None, lookalike=None):
        """สร้าง Targeting"""
        targeting = {
            "geo_locations": {"countries": countries},
            "age_min": age_min,
            "age_max": age_max,
        }
        if genders:
            targeting["genders"] = genders
        if interests:
            targeting["flexible_spec"] = [{"interests": interests}]
        if lookalike:
            targeting["custom_audiences"] = [{"id": lookalike}]
        return targeting

# ตัวอย่าง Campaign Structure
manager = FacebookAdsManager("act_123456", "token")

# Broad Targeting Campaign
broad = CampaignConfig(
    name="[CBO] Broad - Conversions",
    objective=CampaignObjective.CONVERSIONS,
    budget_daily=50.0,
    use_cbo=True,
    ad_sets=[
        AdSetConfig(
            name="Broad 25-55 TH",
            targeting={"countries": ["TH"], "age_min": 25, "age_max": 55},
            budget_daily=0,
            optimization_goal="CONVERSIONS",
            creatives=[
                AdCreative("Image Ad 1", "สินค้าลดราคา 50%", "ช้อปเลย", "SHOP_NOW",
                          format="single_image"),
                AdCreative("Video Ad 1", "รีวิวสินค้ายอดนิยม", "ดูเลย", "SHOP_NOW",
                          format="video"),
            ],
        ),
    ],
)

manager.create_campaign(broad)
print(f"  Ad Sets: {len(broad.ad_sets)}")
for adset in broad.ad_sets:
    print(f"    {adset.name}: {len(adset.creatives)} creatives")

Performance Optimization

# ads_optimizer.py — Facebook Ads Performance Optimization
import random
from dataclasses import dataclass
from typing import List, Dict
from datetime import datetime, timedelta

@dataclass
class AdMetrics:
    ad_name: str
    impressions: int
    clicks: int
    conversions: int
    spend: float
    revenue: float

    @property
    def ctr(self):
        return self.clicks / self.impressions * 100 if self.impressions > 0 else 0

    @property
    def cpc(self):
        return self.spend / self.clicks if self.clicks > 0 else 0

    @property
    def cpa(self):
        return self.spend / self.conversions if self.conversions > 0 else 0

    @property
    def roas(self):
        return self.revenue / self.spend if self.spend > 0 else 0

class AdsOptimizer:
    """Facebook Ads Optimization Engine"""

    def __init__(self):
        self.ads: List[AdMetrics] = []
        self.rules: List[Dict] = []

    def add_ad(self, metrics: AdMetrics):
        self.ads.append(metrics)

    def add_rule(self, name, metric, operator, threshold, action):
        self.rules.append({
            "name": name, "metric": metric,
            "operator": operator, "threshold": threshold,
            "action": action,
        })

    def evaluate_rules(self):
        """ประเมิน Optimization Rules"""
        print(f"\n{'='*55}")
        print(f"Ads Optimization Report")
        print(f"{'='*55}")

        actions = []
        for ad in self.ads:
            for rule in self.rules:
                value = getattr(ad, rule["metric"], None)
                if value is None:
                    continue

                triggered = False
                if rule["operator"] == ">" and value > rule["threshold"]:
                    triggered = True
                elif rule["operator"] == "<" and value < rule["threshold"]:
                    triggered = True

                if triggered:
                    actions.append({
                        "ad": ad.ad_name,
                        "rule": rule["name"],
                        "value": value,
                        "action": rule["action"],
                    })

        if actions:
            print(f"\n  Actions ({len(actions)}):")
            for a in actions:
                print(f"    {a['ad']}: {a['rule']} "
                      f"(value={a['value']:.2f}) -> {a['action']}")
        else:
            print(f"\n  No actions needed")

    def performance_report(self):
        """Performance Report"""
        print(f"\n  Ad Performance:")
        total_spend = sum(a.spend for a in self.ads)
        total_revenue = sum(a.revenue for a in self.ads)

        for ad in sorted(self.ads, key=lambda a: a.roas, reverse=True):
            print(f"    {ad.ad_name:<25} "
                  f"ROAS:{ad.roas:.1f}x "
                  f"CPA: "
                  f"CTR:{ad.ctr:.1f}% "
                  f"Conv:{ad.conversions}")

        print(f"\n  Total: Spend= "
              f"Revenue= "
              f"ROAS={total_revenue/total_spend:.1f}x" if total_spend > 0 else "")

# ตัวอย่าง
optimizer = AdsOptimizer()

ads = [
    AdMetrics("Image Ad - Product A", 50000, 1200, 45, 300, 2700),
    AdMetrics("Video Ad - Review", 80000, 2400, 72, 500, 5400),
    AdMetrics("Carousel - Collection", 30000, 600, 15, 200, 900),
    AdMetrics("UGC Video - Testimonial", 60000, 1800, 90, 400, 6300),
    AdMetrics("Static - Discount 50%", 40000, 800, 8, 250, 400),
]

for ad in ads:
    optimizer.add_ad(ad)

# Optimization Rules
optimizer.add_rule("High CPA", "cpa", ">", 20, "Pause or reduce budget")
optimizer.add_rule("Low ROAS", "roas", "<", 2.0, "Review creative/targeting")
optimizer.add_rule("Low CTR", "ctr", "<", 1.0, "Refresh creative")

optimizer.performance_report()
optimizer.evaluate_rules()

Creative Testing Framework

# creative_testing.py — A/B Testing Framework สำหรับ Ad Creatives
from dataclasses import dataclass
from typing import List
import random
import math

@dataclass
class CreativeVariant:
    name: str
    format: str  # image, video, carousel
    hook: str
    impressions: int = 0
    conversions: int = 0
    spend: float = 0

    @property
    def cvr(self):
        return self.conversions / self.impressions if self.impressions > 0 else 0

class CreativeTestingFramework:
    """Creative Testing Framework ตามแนวคิด Andrew Perpetua"""

    def __init__(self):
        self.variants: List[CreativeVariant] = []
        self.min_impressions = 1000  # Minimum สำหรับ Statistical Significance

    def add_variant(self, variant: CreativeVariant):
        self.variants.append(variant)

    def is_significant(self, v1: CreativeVariant, v2: CreativeVariant):
        """ตรวจสอบ Statistical Significance"""
        if v1.impressions < self.min_impressions or v2.impressions < self.min_impressions:
            return False, "Need more data"

        p1 = v1.cvr
        p2 = v2.cvr
        n1 = v1.impressions
        n2 = v2.impressions

        if p1 == 0 and p2 == 0:
            return False, "No conversions"

        p_pool = (v1.conversions + v2.conversions) / (n1 + n2)
        if p_pool == 0 or p_pool == 1:
            return False, "Cannot calculate"

        se = math.sqrt(p_pool * (1 - p_pool) * (1/n1 + 1/n2))
        if se == 0:
            return False, "SE is zero"

        z = abs(p1 - p2) / se
        return z > 1.96, f"z={z:.2f} (need >1.96)"

    def testing_report(self):
        """Creative Testing Report"""
        print(f"\n{'='*55}")
        print(f"Creative Testing Report")
        print(f"{'='*55}")

        # Sort by CVR
        sorted_variants = sorted(self.variants, key=lambda v: v.cvr, reverse=True)

        winner = sorted_variants[0] if sorted_variants else None

        for v in sorted_variants:
            is_winner = " *WINNER*" if v == winner else ""
            print(f"  {v.name:<30} "
                  f"CVR:{v.cvr*100:.2f}% "
                  f"Conv:{v.conversions} "
                  f"Imp:{v.impressions:,}{is_winner}")

        # Statistical Significance
        if len(sorted_variants) >= 2:
            sig, msg = self.is_significant(sorted_variants[0], sorted_variants[1])
            print(f"\n  Significance: {'YES' if sig else 'NO'} ({msg})")

        # Recommendations
        print(f"\n  Testing Framework (Andrew Perpetua):")
        print(f"    1. ทดสอบ 3-5 Creatives ต่อ Ad Set")
        print(f"    2. รอ 1,000+ Impressions ก่อนตัดสินใจ")
        print(f"    3. ดู CVR ไม่ใช่แค่ CTR")
        print(f"    4. Winner takes all — Scale winner, kill losers")
        print(f"    5. ทดสอบ Creative ใหม่ทุก 2 สัปดาห์")

# ตัวอย่าง
framework = CreativeTestingFramework()

variants = [
    CreativeVariant("UGC Video - Unboxing", "video", "ลองเปิดกล่อง!", 15000, 120, 300),
    CreativeVariant("Studio Photo - Product", "image", "สินค้าขายดี", 12000, 72, 240),
    CreativeVariant("Carousel - Features", "carousel", "5 เหตุผลที่ต้องมี", 10000, 50, 200),
    CreativeVariant("Meme Style - Humor", "image", "เมื่อเพื่อนเห็นของ", 8000, 40, 160),
    CreativeVariant("Before/After", "image", "ก่อน vs หลัง", 11000, 88, 220),
]

for v in variants:
    framework.add_variant(v)

framework.testing_report()

เคล็ดลับจาก Andrew Perpetua

• Broad Targeting: ให้ Algorithm หา Audience เอง อย่า Narrow มากเกินไป
• CBO: ใช้ Campaign Budget Optimization ให้ Facebook จัดสรร Budget อัตโนมัติ
• Creative First: Creative สำคัญที่สุด ทดสอบหลาย Format (Video, Image, UGC)
• ROAS Focus: วัดผลด้วย ROAS ไม่ใช่แค่ CTR หรือ CPC
• Scale Gradually: เพิ่ม Budget 20% ต่อสัปดาห์ อย่าเพิ่มทีเดียวมาก
• Advantage+: ใช้ Advantage+ Shopping Campaigns สำหรับ E-commerce

Andrew Perpetua คือใคร

ผู้เชี่ยวชาญ Facebook Ads Digital Marketing เคยทำงานที่ Meta Product Marketing Manager ดูแล Ads Products ประสบการณ์ 10+ ปี Performance Marketing แชร์ความรู้ Blog Social Media

สรุป

Andrew Perpetua สอนแนวคิด Facebook Ads ที่ Data-driven Broad Targeting ให้ Algorithm ทำงาน CBO จัดสรร Budget Creative สำคัญที่สุด ทดสอบหลาย Format วัดผล ROAS Scale ค่อยๆเพิ่ม Budget Advantage+ สำหรับ E-commerce