Emergency News Agent - 긴급 뉴스 모니터

Role

Grounding API를 활용하여 시장 영향력이 큰 긴급 뉴스를 실시간 감지하고 즉각 알림합니다.

Core Capabilities

1. Grounding API Integration

from anthropic import Anthropic

anthropic = Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))

async def monitor_breaking_news(query: str) -> List[Dict]: """Monitor breaking news using Grounding API"""

message = anthropic.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": query
    }],
    tools=[{
        "type": "web_search_20241022",
        "name": "web_search",
        "search_query": "breaking market news stock",
        "max_results": 10
    }]
)

# Parse results
search_results = extract_search_results(message)

return search_results

2. Urgency Classification

def classify_urgency(news: Dict) -> str: """Classify news urgency based on content"""

# CRITICAL keywords
critical_keywords = [
    "halt", "suspended", "emergency",
    "bankruptcy", "sec investigation",
    "fda rejection", "recalls"
]

# HIGH keywords
high_keywords = [
    "acquisition", "merger",
    "earnings miss", "guidance cut",
    "ceo resignation", "lawsuit"
]

# MEDIUM keywords
medium_keywords = [
    "partnership", "new product",
    "analyst upgrade", "contract win"
]

headline = news['headline'].lower()
content = news.get('content', '').lower()
text = headline + ' ' + content

# Check CRITICAL
if any(kw in text for kw in critical_keywords):
    return "CRITICAL"

# Check HIGH
if any(kw in text for kw in high_keywords):
    return "HIGH"

# Check MEDIUM
if any(kw in text for kw in medium_keywords):
    return "MEDIUM"

return "LOW"

3. Ticker Extraction

import re

def extract_tickers(text: str) -> List[str]: """Extract stock tickers from text"""

# Common patterns
patterns = [
    r'\b([A-Z]{1,5})\b',  # All caps 1-5 letters
    r'\$([A-Z]{1,5})\b',  # $AAPL format
    r'NYSE:\s*([A-Z]{1,5})',  # NYSE: AAPL
    r'NASDAQ:\s*([A-Z]{1,5})'  # NASDAQ: AAPL
]

tickers = set()

for pattern in patterns:
    matches = re.findall(pattern, text)
    tickers.update(matches)

# Filter out common false positives
false_positives = {'CEO', 'FDA', 'SEC', 'USA', 'IPO', 'ETF'}
tickers = tickers - false_positives

# Validate against known tickers
valid_tickers = [t for t in tickers if is_valid_ticker(t)]

return valid_tickers

4. Impact Assessment

def assess_impact( urgency: str, ticker: str, news_type: str ) -> Dict: """Assess potential market impact"""

# Base impact by urgency
BASE_IMPACT = {
    "CRITICAL": 0.15,    # ±15%
    "HIGH": 0.08,        # ±8%
    "MEDIUM": 0.03,      # ±3%
    "LOW": 0.01          # ±1%
}

# News type multiplier
TYPE_MULTIPLIER = {
    "fda_approval": 1.5,
    "fda_rejection": 1.8,
    "bankruptcy": 2.0,
    "merger": 1.3,
    "earnings": 1.0,
    "partnership": 0.8
}

base = BASE_IMPACT[urgency]
multiplier = TYPE_MULTIPLIER.get(news_type, 1.0)

estimated_impact = base * multiplier

# Determine direction
direction = infer_direction(news_type)

return {
    "estimated_price_impact": estimated_impact,
    "direction": direction,  # POSITIVE/NEGATIVE
    "confidence": 0.6,
    "timeframe": "immediate" if urgency == "CRITICAL" else "short_term"
}

5. Alert Triggering

async def trigger_alert(news: Dict, urgency: str): """Trigger appropriate alerts based on urgency"""

if urgency == "CRITICAL":
    # Immediate actions
    await send_telegram_alert(news, priority="URGENT")
    await notify_war_room(news)
    await broadcast_websocket(news)
    
    # Auto-trigger War Room debate
    if news.get('tickers'):
        for ticker in news['tickers']:
            await initiate_emergency_debate(ticker, news)

elif urgency == "HIGH":
    await send_telegram_alert(news, priority="HIGH")
    await broadcast_websocket(news)

elif urgency == "MEDIUM":
    await broadcast_websocket(news)

# Always save to database
save_emergency_news(news)

Decision Framework

Step 1: Continuous Monitoring (Every 60 seconds) search_query = "breaking stock market news"

results = grounding_api.search( query=search_query, recency="1_hour", max_results=10 )

Step 2: Filter New Articles FOR each result in results: IF not exists_in_db(result.url): new_articles.append(result)

Step 3: Classify Urgency FOR article in new_articles: urgency = classify_urgency(article)

IF urgency == "LOW":
  SKIP (not market-moving)

Step 4: Extract Tickers tickers = extract_tickers(article.content)

IF len(tickers) == 0: tickers = ["SPY"] # Market-wide news

Step 5: Assess Impact FOR ticker in tickers: impact = assess_impact(urgency, ticker, news_type)

Step 6: Create NewsArticle Entry article_entry = NewsArticle( ticker=ticker, headline=article.headline, content=article.content, source="emergency_news", urgency=urgency, sentiment_score=calculate_sentiment(article), created_at=datetime.now() )

db.add(article_entry)

Step 7: Trigger Alerts await trigger_alert(article, urgency)

Step 8: IF CRITICAL:

Auto-initiate War Room debate

await initiate_emergency_debate(ticker, article)

Output Format

{ "alert_id": "EMERG-20251221-001", "timestamp": "2025-12-21T13:05:23Z", "urgency": "CRITICAL", "detection_latency_sec": 45,

"news_details": { "headline": "FDA Rejects Moderna Cancer Vaccine - Major Setback", "summary": "FDA citing safety concerns in Phase 3 trial data. Moderna stock halted.", "source": "Reuters", "url": "https://reuters.com/article/...", "published_at": "2025-12-21T13:04:38Z" },

"affected_tickers": [ { "ticker": "MRNA", "relationship": "primary", "impact_assessment": { "estimated_price_impact": -0.27, "direction": "NEGATIVE", "confidence": 0.85, "timeframe": "immediate", "reasoning": "FDA 거부는 신약 파이프라인 붕괴, 역사적으로 -20~-30% 급락" } }, { "ticker": "PFE", "relationship": "competitor", "impact_assessment": { "estimated_price_impact": 0.08, "direction": "POSITIVE", "confidence": 0.60, "timeframe": "short_term", "reasoning": "경쟁사 Moderna 약세는 PFE에 긍정적" } } ],

"urgency_classification": { "level": "CRITICAL", "confidence": 0.95, "reasoning": "FDA rejection + trading halt = 시장 즉각 반응", "keywords_matched": ["fda rejection", "halted", "safety concerns"] },

"actions_taken": [ { "action": "telegram_alert_sent", "recipient": "COMMANDER", "timestamp": "2025-12-21T13:05:25Z", "priority": "URGENT" }, { "action": "war_room_debate_initiated", "ticker": "MRNA", "timestamp": "2025-12-21T13:05:26Z" }, { "action": "websocket_broadcast", "connections_notified": 5, "timestamp": "2025-12-21T13:05:24Z" }, { "action": "database_entry_created", "news_id": 1024, "timestamp": "2025-12-21T13:05:23Z" } ],

"recommended_actions": [ "Review MRNA positions immediately", "Consider PFE as alternative", "Monitor for official company response" ] }

Examples

Example 1: CRITICAL - Trading Halt

Detected: "Tesla trading halted pending SEC investigation"

Actions:

1. TELEGRAM → Commander (URGENT)
2. War Room → Emergency debate on TSLA
3. WebSocket → All connected clients
4. Database → Save with urgency=CRITICAL

Expected Impact: -15% to -25%

Example 2: HIGH - Major Acquisition

Detected: "Microsoft to acquire OpenAI for $80B"

Actions:

1. TELEGRAM → Commander (HIGH priority)
2. WebSocket → Broadcast
3. Database → Save with urgency=HIGH

Expected Impact: MSFT +8%, GOOGL -3%

Example 3: MEDIUM - Partnership

Detected: "Apple partners with Samsung on chip development"

Actions:

1. WebSocket → Broadcast
2. Database → Save with urgency=MEDIUM

Expected Impact: AAPL +2%, Samsung +1%

Guidelines

Do's ✅

• 속도 최우선: Detection latency < 2분

• False Positive 허용: 놓치는 것보다 나음 (Recall > Precision)

• 즉각 알림: CRITICAL은 모든 채널 동시 알림

• War Room 자동 촉발: CRITICAL 뉴스는 자동 debate

Don'ts ❌

• 과도한 필터링 금지 (중요 뉴스 놓칠 위험)

• Grounding API 남용 금지 (rate limit 준수)

• CRITICAL 남발 금지 (신뢰도 유지)

• Latency > 5분 금지 (긴급성 손실)

Integration

Continuous Monitoring Loop

import asyncio

async def emergency_news_monitor(): """Continuous monitoring loop"""

logger.info("Emergency News Monitor started")

while True:
    try:
        # Search for breaking news
        results = await monitor_breaking_news(
            query="breaking stock market news OR urgent company announcement"
        )
        
        # Process results
        for result in results:
            # Check if already processed
            if not is_duplicate(result):
                await process_emergency_news(result)
        
        # Sleep for 60 seconds
        await asyncio.sleep(60)
    
    except Exception as e:
        logger.error(f"Emergency monitor error: {e}")
        await asyncio.sleep(10)  # Shorter retry on error

Startup Integration

backend/main.py

from backend.ai.skills.analysis.emergency_news_agent import emergency_news_monitor

@app.on_event("startup") async def startup_event(): # Start emergency news monitor in background asyncio.create_task(emergency_news_monitor()) logger.info("Emergency News Agent activated")

War Room Auto-Trigger

async def initiate_emergency_debate(ticker: str, news: Dict): """Auto-trigger War Room debate for critical news"""

from backend.ai.debate.ai_debate_engine import AIDebateEngine

debate_engine = AIDebateEngine()

# Run emergency debate
result = await debate_engine.run_debate(
    ticker=ticker,
    emergency_mode=True,
    context={
        "news_headline": news['headline'],
        "urgency": "CRITICAL",
        "detection_time": datetime.now()
    }
)

# Send result to Commander
await send_telegram_message(
    f"🚨 Emergency Debate Result for {ticker}\n\n"
    f"Decision: {result['final_decision']}\n"
    f"Confidence: {result['final_confidence']:.0%}\n"
    f"News: {news['headline']}"
)

Performance Metrics

• Detection Latency: 목표 < 2분 (뉴스 발생 → 알림)

• Recall: > 95% (중요 뉴스 놓치지 않기)

• Precision: > 60% (False positive 허용)

• Uptime: > 99.9% (24/7 모니터링)

Alert Examples

Telegram Alert Format

🚨 CRITICAL ALERT 🚨

Ticker: MRNA Impact: -27% (immediate)

FDA REJECTS CANCER VACCINE

Summary: FDA citing safety concerns. Trading halted.

Actions Taken: ✅ War Room debate initiated ✅ All systems notified

Recommendation: Review positions NOW

Time: 13:05:23 UTC Latency: 45 seconds

Version History

• v1.0 (2025-12-21): Initial release with Grounding API and auto-alert system