langfuse-fred-analysis.py

def get_fred_series(series_id: str) -> Optional[Dict[str, Any]]:
    """
    Fetches latest observations for a FRED series.
    """
    fred_api_key = os.getenv("FRED_API_KEY") or get_parameter("fred_api_key")
    if not fred_api_key:
        logger.error("Missing FRED API key.")
        return None

    url = "https://api.stlouisfed.org/fred/series/observations"
    params = {
        "series_id": series_id,
        "api_key": fred_api_key,
        "file_type": "json",
        "sort_order": "desc",
        "limit": 10
    }

    try:
        resp = requests.get(url, params=params)
        resp.raise_for_status()
        return resp.json()
    except Exception as exc:
        logger.error("FRED fetch error [%s]: %s", series_id, exc)
        return None


def construct_prompt(series_id: str, date: str, value: str, history: List[Dict[str, str]]) -> str:
    """
    Builds prompt for financial analysis based on latest and historical data.
    """
    return (
        f"Analyze FRED series {series_id} (Real GDP), latest value: {value} on {date}.\n\n"
        "1. Trend Comparison: Highlight recent vs. long-term changes...\n"
        "2. Risk Assessment: Flag anomalies, explain deviations...\n"
        "3. External Factors: Consider fiscal/monetary policy, global events...\n"
        "4. Forecast: Project baseline/worst/best-case scenarios...\n"
        "5. Actionable Insights: Suggest investment/strategic moves...\n\n"
        "Note data limitations (seasonal lags, etc.), compare to international benchmarks.\n"
        "Structure your response like a seasoned financial analyst."
    )


def analyze_financial_data(trace: Any, series_id: str = "GDPC1") -> str:
    """
    Retrieves FRED data and prompts LLM for analysis.
    """
    fred_data = get_fred_series(series_id)
    observations = fred_data.get("observations", []) if fred_data else []

    if not observations:
        log_event(trace, "fred_error", {"series_id": series_id, "error": "No data"})
        return "Error: Unable to retrieve FRED data."

    latest = observations[0]
    date, value = latest.get("date", "Unknown"), latest.get("value", "Unknown")
    history = [{"date": obs["date"], "value": obs["value"]} for obs in observations[1:]]

    logger.info("FRED %s | %s: %s", series_id, date, value)
    logger.debug("Historical points: %d", len(history))

    prompt = construct_prompt(series_id, date, value, history)
    log_event(trace, "prompt_sent", {"message": "Prompt sent"}, input_data={"prompt": prompt})
    logger.info("Prompt:\n%s", prompt)

    start = time.time()
    analysis = get_model_response(prompt)
    latency = int((time.time() - start) * 1000)

    log_event(trace, "response_received", {"latency_ms": latency, "model": "OpenAI GPT-4"},
              output_data={"response": analysis})
    return analysis