pranjalAI

def normalize_ipo_calendar(ipo_df: pd.DataFrame) -> pd.DataFrame:
    df = ipo_df.copy()
    if df.empty:
        return df

    # Defensive: use `.get()` so missing columns don't raise KeyError.
    df["symbol"] = df.get("symbol")
    df["companyName"] = df.get("companyName")
    df["exchange"] = df.get("exchange")
    df["ipoDate"] = pd.to_datetime(df.get("date"), errors="coerce")

pranjalAI

{
  "symbol": "AMD",
  "label": "Needs Review",
  "signals_fired": 2,
  "signals": [
    {
      "name": "ClusterActivity",
      "value": 4,
      "fired": true,
      "why": "4 unique insiders traded in the last 30 days."

pranjalAI

@dataclass
class NarrativeAgent:
    def run(self, state: Dict[str, Any]) -> Dict[str, Any]:
        symbol = state["symbol"]
        signals = state.get("signals", [])

        fired = [s for s in signals if s.get("fired")]
        score = len(fired)

        if score >= 3:

pranjalAI

# agents/ratios_agent.py
from state import ResearchState
from mcp_client import call_mcp_tool

class RatiosAgent:
    def run(self, state: ResearchState) -> ResearchState:
        symbol = state.symbol
        if not symbol:
            raise ValueError("Symbol missing in state.")

pranjalAI

# agents/fundamentals_agent.py
from state import ResearchState
from mcp_client import call_mcp_tool

class FundamentalsAgent:
    def run(self, state: ResearchState) -> ResearchState:
        symbol = state.symbol
        if not symbol:
            raise ValueError("Symbol missing in state.")

pranjalAI

# agents/profile_agent.py
from typing import Optional
from state import ResearchState
from mcp_client import call_mcp_tool

class ProfileAgent:
    def _extract_symbol(self, query: str) -> Optional[str]:
        # simple heuristic: if user passes "NVDA" or "AAPL"
        parts = query.strip().split()
        for token in parts:

pranjalAI

from agents.profile_agent import ProfileAgent
from agents.fundamentals_agent import FundamentalsAgent
from agents.ratios_agent import RatiosAgent
from agents.risk_agent import RiskAgent
from agents.summary_agent import SummaryAgent
from state import ResearchState

def run_research_pipeline(query: str) -> ResearchState:
    state = ResearchState(query=query)

pranjalAI

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

@dataclass
class ResearchState:
    # user input
    query: str
    symbol: Optional[str] = None

    # core data blobs

pranjalAI

import requests
from config import FMP_MCP_ENDPOINT

def call_mcp_tool(tool_name: str, params: dict | None = None):
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": tool_name,
        "params": params or {}
    }

pranjalAI

mylst = [10,20,30]
a,b,c = mylst
print(c, a, b )
print(a, c, b )