254 lines
8.2 KiB
Python
Executable File
254 lines
8.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
ANALYZE CREATIVE TEST RESULTS
|
|
Analyse les résultats des tests et sélectionne les winners par ISP
|
|
"""
|
|
|
|
import psycopg2
|
|
from datetime import datetime, timedelta
|
|
|
|
DB_CONFIG = {
|
|
'host': 'localhost',
|
|
'database': 'adx_system',
|
|
'user': 'admin',
|
|
'password': 'admin123'
|
|
}
|
|
|
|
def get_db():
|
|
return psycopg2.connect(**DB_CONFIG)
|
|
|
|
def analyze_results(conn, batch_id=None):
|
|
"""Analyze test results"""
|
|
cur = conn.cursor()
|
|
|
|
# Get results by creative and ISP
|
|
query = """
|
|
SELECT
|
|
t.offer_id,
|
|
t.subject_line,
|
|
t.recipient_isp,
|
|
COUNT(*) as sent,
|
|
SUM(CASE WHEN t.opened THEN 1 ELSE 0 END) as opens,
|
|
SUM(CASE WHEN t.clicked THEN 1 ELSE 0 END) as clicks,
|
|
SUM(CASE WHEN t.bounced THEN 1 ELSE 0 END) as bounces
|
|
FROM admin.brain_warmup_tests t
|
|
WHERE t.config_type = 'creative_test'
|
|
"""
|
|
|
|
if batch_id:
|
|
query += f" AND t.test_batch_id = '{batch_id}'"
|
|
else:
|
|
query += " AND t.sent_at > NOW() - INTERVAL '7 days'"
|
|
|
|
query += " GROUP BY t.offer_id, t.subject_line, t.recipient_isp ORDER BY t.offer_id, clicks DESC"
|
|
|
|
cur.execute(query)
|
|
return cur.fetchall()
|
|
|
|
def calculate_scores(conn):
|
|
"""Calculate and update creative scores"""
|
|
cur = conn.cursor()
|
|
|
|
# Get aggregated stats per creative
|
|
cur.execute("""
|
|
WITH stats AS (
|
|
SELECT
|
|
c.id as creative_id,
|
|
COUNT(t.id) as total_sent,
|
|
SUM(CASE WHEN t.opened THEN 1 ELSE 0 END) as total_opens,
|
|
SUM(CASE WHEN t.clicked THEN 1 ELSE 0 END) as total_clicks,
|
|
SUM(CASE WHEN t.bounced THEN 1 ELSE 0 END) as total_bounces
|
|
FROM admin.offer_creatives c
|
|
LEFT JOIN admin.brain_warmup_tests t ON t.subject_line = c.subject_line
|
|
WHERE t.config_type = 'creative_test'
|
|
GROUP BY c.id
|
|
)
|
|
UPDATE admin.offer_creatives c
|
|
SET
|
|
opens = COALESCE(s.total_opens, 0),
|
|
clicks = COALESCE(s.total_clicks, 0),
|
|
times_used = COALESCE(s.total_sent, times_used),
|
|
open_rate = CASE WHEN s.total_sent > 0 THEN (s.total_opens::decimal / s.total_sent * 100) ELSE NULL END,
|
|
click_rate = CASE WHEN s.total_opens > 0 THEN (s.total_clicks::decimal / s.total_opens * 100) ELSE NULL END,
|
|
score = CASE
|
|
WHEN COALESCE(s.total_sent, 0) = 0 THEN 50
|
|
WHEN COALESCE(s.total_sent, 0) < 10 THEN 50 + COALESCE(s.total_clicks, 0) * 5
|
|
ELSE LEAST(100,
|
|
(COALESCE(s.total_opens::decimal / NULLIF(s.total_sent, 0), 0) * 100 * 0.3) +
|
|
(COALESCE(s.total_clicks::decimal / NULLIF(s.total_opens, 0), 0) * 100 * 0.7)
|
|
)
|
|
END
|
|
FROM stats s
|
|
WHERE c.id = s.creative_id
|
|
""")
|
|
conn.commit()
|
|
|
|
def select_winners(conn):
|
|
"""Select winner creative per offer"""
|
|
cur = conn.cursor()
|
|
|
|
# Reset all winners
|
|
cur.execute("UPDATE admin.offer_creatives SET is_winner = false")
|
|
|
|
# Select best per offer
|
|
cur.execute("""
|
|
UPDATE admin.offer_creatives c
|
|
SET is_winner = true
|
|
WHERE c.id IN (
|
|
SELECT DISTINCT ON (offer_id) id
|
|
FROM admin.offer_creatives
|
|
WHERE status = 'active'
|
|
ORDER BY offer_id, score DESC, clicks DESC, times_used ASC
|
|
)
|
|
""")
|
|
conn.commit()
|
|
|
|
def get_winners_by_isp(conn):
|
|
"""Get best creative per offer per ISP"""
|
|
cur = conn.cursor()
|
|
|
|
cur.execute("""
|
|
WITH isp_stats AS (
|
|
SELECT
|
|
t.offer_id,
|
|
c.id as creative_id,
|
|
c.subject_line,
|
|
t.recipient_isp,
|
|
COUNT(*) as sent,
|
|
SUM(CASE WHEN t.clicked THEN 1 ELSE 0 END) as clicks,
|
|
CASE WHEN COUNT(*) > 0
|
|
THEN (SUM(CASE WHEN t.clicked THEN 1 ELSE 0 END)::decimal / COUNT(*) * 100)
|
|
ELSE 0
|
|
END as click_rate
|
|
FROM admin.brain_warmup_tests t
|
|
JOIN admin.offer_creatives c ON t.subject_line = c.subject_line AND t.offer_id = c.offer_id
|
|
WHERE t.config_type = 'creative_test'
|
|
GROUP BY t.offer_id, c.id, c.subject_line, t.recipient_isp
|
|
)
|
|
SELECT
|
|
offer_id,
|
|
recipient_isp,
|
|
creative_id,
|
|
subject_line,
|
|
sent,
|
|
clicks,
|
|
ROUND(click_rate::numeric, 2) as click_rate
|
|
FROM isp_stats
|
|
WHERE (offer_id, recipient_isp, click_rate) IN (
|
|
SELECT offer_id, recipient_isp, MAX(click_rate)
|
|
FROM isp_stats
|
|
GROUP BY offer_id, recipient_isp
|
|
)
|
|
ORDER BY offer_id, recipient_isp
|
|
""")
|
|
return cur.fetchall()
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print("📊 CREATIVE RESULTS ANALYZER")
|
|
print(f"Time: {datetime.now()}")
|
|
print("=" * 60)
|
|
|
|
conn = get_db()
|
|
|
|
# Analyze all test results
|
|
print("\n🔍 Analyzing test results...")
|
|
results = analyze_results(conn)
|
|
|
|
if not results:
|
|
print("⚠️ No test results found. Run test-creatives.py first.")
|
|
return
|
|
|
|
print(f"Found {len(results)} result rows")
|
|
|
|
# Calculate scores
|
|
print("\n📈 Calculating scores...")
|
|
calculate_scores(conn)
|
|
|
|
# Select winners
|
|
print("\n🏆 Selecting winners...")
|
|
select_winners(conn)
|
|
|
|
# Show results
|
|
print("\n" + "=" * 60)
|
|
print("📋 RESULTS BY CREATIVE")
|
|
print("=" * 60)
|
|
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
SELECT
|
|
c.id,
|
|
LEFT(c.subject_line, 35) as subject,
|
|
c.times_used as sent,
|
|
c.opens,
|
|
c.clicks,
|
|
ROUND(c.open_rate::numeric, 1) as open_pct,
|
|
ROUND(c.click_rate::numeric, 1) as click_pct,
|
|
ROUND(c.score::numeric, 1) as score,
|
|
c.is_winner as win
|
|
FROM admin.offer_creatives c
|
|
WHERE c.status = 'active'
|
|
ORDER BY c.score DESC
|
|
""")
|
|
|
|
print(f"{'ID':>3} | {'Subject':<35} | {'Sent':>4} | {'Open':>4} | {'Click':>5} | {'O%':>5} | {'C%':>5} | {'Score':>5} | Win")
|
|
print("-" * 95)
|
|
|
|
for row in cur.fetchall():
|
|
cid, subj, sent, opens, clicks, open_pct, click_pct, score, win = row
|
|
win_mark = "✅" if win else ""
|
|
print(f"{cid:>3} | {subj:<35} | {sent or 0:>4} | {opens or 0:>4} | {clicks or 0:>5} | {open_pct or 0:>5} | {click_pct or 0:>5} | {score or 0:>5} | {win_mark}")
|
|
|
|
# Winners by ISP
|
|
print("\n" + "=" * 60)
|
|
print("🏆 BEST CREATIVE BY ISP")
|
|
print("=" * 60)
|
|
|
|
isp_winners = get_winners_by_isp(conn)
|
|
|
|
current_offer = None
|
|
for offer_id, isp, cid, subject, sent, clicks, click_rate in isp_winners:
|
|
if offer_id != current_offer:
|
|
print(f"\nOffer #{offer_id}:")
|
|
current_offer = offer_id
|
|
print(f" {isp:<10}: #{cid} {subject[:30]}... ({clicks}/{sent} clicks, {click_rate}%)")
|
|
|
|
# Recommendations
|
|
print("\n" + "=" * 60)
|
|
print("💡 RECOMMENDATIONS")
|
|
print("=" * 60)
|
|
|
|
cur.execute("""
|
|
SELECT c.id, c.subject_line, c.score, c.clicks, c.times_used
|
|
FROM admin.offer_creatives c
|
|
WHERE c.is_winner = true AND c.score >= 50 AND c.times_used >= 5
|
|
ORDER BY c.score DESC
|
|
""")
|
|
|
|
ready = cur.fetchall()
|
|
if ready:
|
|
print("\n✅ READY FOR MASS SEND (tested & performing):")
|
|
for cid, subj, score, clicks, used in ready:
|
|
print(f" Creative #{cid}: {subj[:40]}... (score: {score}, {clicks} clicks)")
|
|
else:
|
|
print("\n⚠️ No creatives ready yet. Need more test data.")
|
|
print(" Run: python3 /opt/wevads/scripts/test-creatives.py 10")
|
|
|
|
cur.execute("""
|
|
SELECT c.id, c.subject_line, c.times_used
|
|
FROM admin.offer_creatives c
|
|
WHERE c.times_used < 5 AND c.status = 'active'
|
|
""")
|
|
|
|
need_test = cur.fetchall()
|
|
if need_test:
|
|
print(f"\n⏳ NEED MORE TESTING ({len(need_test)} creatives):")
|
|
for cid, subj, used in need_test[:5]:
|
|
print(f" Creative #{cid}: {subj[:40]}... (only {used} sends)")
|
|
|
|
conn.close()
|
|
print("\n✅ Analysis complete!")
|
|
|
|
if __name__ == '__main__':
|
|
main()
|