📄 premarket_briefing.py 5,194 bytes May 01, 2026 📋 Raw

!/usr/bin/env python3

"""
Pre-market briefing job (AM).
Runs at 8:00 AM CT before market open.
"""
import sys
import os
from datetime import datetime

Add workspace to path

sys.path.insert(0, "/home/hoffmann_admin/.openclaw/workspace")

from core.market import (
get_watchlist_with_metadata,
get_top_stories,
get_insider_sentiment_summary,
get_earnings_this_week,
format_news_headline,
check_staleness,
should_skip_briefing,
load_user_watchlist
)
from core.market.news import extract_top_headlines

def format_price(price: float, change: float, change_pct: float) -> str:
"""Format price with change indicator."""
sign = "+" if change >= 0 else ""
emoji = "🟢" if change >= 0 else "🔴"
return f"${price:.2f} ({sign}{change:.2f} / {sign}{change_pct:.2f}%) {emoji}"

def generate_premarket_briefing() -> str:
"""Generate the pre-market briefing."""
today = datetime.now().strftime("%A, %B %d, %Y")

# Check for stale data
stale = check_staleness()
stale_warning = ""
if any(stale.values()):
    stale_warning = "⚠️ Some data may be stale\n"

# Check if we should skip (critical data too old)
if should_skip_briefing():
    return f"⚠️ Market Briefing Skipped\n\nData is too stale to provide reliable briefing.\nLast successful fetch: check logs."

# Get watchlist data
watchlist_data = get_watchlist_with_metadata()
user_tickers = watchlist_data["user"]
dynamic_tickers = watchlist_data["dynamic"]
all_symbols = watchlist_data["all_symbols"]

# Build watchlist section
watchlist_lines = []

# User tickers
for ticker in user_tickers:
    symbol = ticker["symbol"]
    price_str = format_price(
        ticker["price"],
        ticker["change"],
        ticker["change_pct"]
    )

    # Get news for this ticker
    from core.market.news import fetch_ticker_news, extract_top_headlines
    news = fetch_ticker_news(symbol)
    top_news = extract_top_headlines(news, 1)

    news_str = ""
    if top_news:
        news_str = f"\n    {format_news_headline(symbol, top_news[0])}"

    watchlist_lines.append(f"• {symbol}: {price_str}{news_str}")

# Dynamic tickers with reasons
for ticker in dynamic_tickers:
    symbol = ticker["symbol"]
    price_str = format_price(
        ticker["price"],
        ticker["change"],
        ticker["change_pct"]
    )
    reason = ticker.get("reason", "Trending")

    # Get news for this ticker
    from core.market.news import fetch_ticker_news, extract_top_headlines
    news = fetch_ticker_news(symbol)
    top_news = extract_top_headlines(news, 1)

    news_str = ""
    if top_news:
        news_str = f"\n    {format_news_headline(symbol, top_news[0])}"

    watchlist_lines.append(f"• {symbol}: {price_str} {reason}{news_str}")

# Get top stories across watchlist
top_stories = get_top_stories(all_symbols, count=5)
stories_lines = []
for story in top_stories[:3]:  # Top 3
    ticker = story.get("_ticker", "")
    headline = story.get("headline", "")
    source = story.get("source", "Unknown")
    if len(headline) > 90:
        headline = headline[:87] + "..."
    stories_lines.append(f"• [{ticker}] {headline} ({source})")

if not stories_lines:
    stories_lines.append("• No major market-moving news today")

# Get insider sentiment
insider = get_insider_sentiment_summary(all_symbols)

# Get earnings
earnings = get_earnings_this_week(all_symbols)

# Build the briefing
briefing = f"""🌅 Pre-Market Brief — {today}

{stale_warning}
📊 Watchlist:
{chr(10).join(watchlist_lines)}

📰 Top Stories:
{chr(10).join(stories_lines)}

🎯 Insider Sentiment:
{chr(10).join(insider)}

📅 This Week:
{chr(10).join(earnings)}
"""

return briefing

def main():
"""Run the pre-market briefing."""
try:
briefing = generate_premarket_briefing()
print(briefing)

    # Send to Telegram if configured
    chat_id = os.environ.get("MARKET_BRIEFING_CHAT_ID", "8386527252")
    if chat_id:
        try:
            # Try to use Telegram API if available
            import requests
            bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
            if bot_token and len(briefing) < 4000:
                url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
                payload = {
                    "chat_id": chat_id,
                    "text": briefing,
                    "parse_mode": "HTML"
                }
                requests.post(url, json=payload, timeout=30)
        except Exception as e:
            print(f"Failed to send Telegram message: {e}")

    return 0
except Exception as e:
    print(f"Error generating briefing: {e}")
    import traceback
    traceback.print_exc()
    return 1

if name == "main":
sys.exit(main())

← Back