import os, requests, time, csv, io
AV_KEY = os.environ["ALPHA_VANTAGE_KEY"]
AV_BASE = "https://www.alphavantage.co/query"
MV = os.environ["MAVERA_API_KEY"]
MV_BASE = "https://app.mavera.io/api/v1"
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
AUDIENCE_SECTORS = ["Technology", "Communication Services", "Consumer Discretionary"]
BRAND = "B2B marketing platform. Audience: Marketing leaders at enterprise companies. Tone: analytical, forward-looking."
# 1. Fetch earnings calendar (returns CSV)
r = requests.get(AV_BASE, params={
"function": "EARNINGS_CALENDAR", "horizon": "3month", "apikey": AV_KEY,
})
r.raise_for_status()
reader = csv.DictReader(io.StringIO(r.text))
all_earnings = [row for row in reader]
print(f"Total earnings events: {len(all_earnings)}")
# 2. Filter to relevant companies
relevant = [e for e in all_earnings
if any(s.lower() in (e.get("name","") + e.get("symbol","")).lower() for s in ["AAPL","MSFT","GOOGL","META","AMZN","SHOP","HBS","CRM","ADBE","TWLO"])
or float(e.get("estimate","0") or "0") > 1.0]
relevant.sort(key=lambda x: x.get("reportDate",""))
print(f"Relevant earnings: {len(relevant)}")
# 3. Build earnings timeline
timeline = "\n".join(
f"- {e.get('reportDate','')} | {e.get('symbol','')} ({e.get('name','')}) | "
f"Est EPS: ${e.get('estimate','N/A')} | Fiscal: {e.get('fiscalDateEnding','')}"
for e in relevant[:20]
)
# 4. Mave content planning
plan = requests.post(f"{MV_BASE}/mave/chat", headers=MV_H, json={
"message": f"Content strategist specializing in earnings-cycle timing.\n\n"
f"BRAND: {BRAND}\n\n"
f"UPCOMING EARNINGS:\n{timeline}\n\n"
"For the top 8 most impactful earnings events, create a content timing plan:\n\n"
"PRE-EARNINGS (3-5 days before):\n- Thought piece angle that positions us as analysts\n"
"EARNINGS DAY:\n- Real-time commentary template (LinkedIn post format)\n"
"POST-EARNINGS (1-2 days after):\n- Analysis framework content piece\n\n"
"Include: publish date, format, headline, key hook, why our audience cares."
}).json()
print(f"\n{plan.get('content', '')[:2000]}")
# 5. Generate first content piece
if relevant:
first = relevant[0]
gen = requests.post(f"{MV_BASE}/generations", headers=MV_H, json={
"prompt": f"Write a pre-earnings thought piece (600 words) for marketing leaders.\n\n"
f"COMPANY: {first.get('name','')} ({first.get('symbol','')})\n"
f"REPORT DATE: {first.get('reportDate','')}\n"
f"EPS ESTIMATE: ${first.get('estimate','N/A')}\n"
f"BRAND: {BRAND}\n\n"
"Angle: What this company's earnings signal about marketing technology spend.\n"
"Include: What to watch, implications for marketing budgets, our unique take.\n"
"Tone: analytical, not financial advice. End with a question for engagement.",
}).json()
print(f"\n{'='*60}\nPRE-EARNINGS PIECE: {first.get('symbol','')}\n{'='*60}")
print(gen.get("output", gen.get("content", ""))[:1000])