langchain_tool.py

"""
LangChain tool for Headless Oracle V5.

Usage:
    from integrations.langchain_tool import MarketStatusTool, MarketScheduleTool
    from langchain.agents import initialize_agent, AgentType
    from langchain_openai import ChatOpenAI

    tools = [MarketStatusTool(api_key="ok_live_..."), MarketScheduleTool()]
    agent = initialize_agent(tools, ChatOpenAI(), agent=AgentType.OPENAI_FUNCTIONS)
    agent.run("Is NYSE open right now? Should I execute this trade?")

Requires:
    pip install "headless-oracle[langchain]"
"""

from __future__ import annotations

from typing import Optional, Type

from langchain_core.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from pydantic import BaseModel, Field

from headless_oracle import OracleClient, verify


class _MicInput(BaseModel):
    mic: str = Field(
        default="XNYS",
        description="Exchange MIC code. Supported: XNYS, XNAS, XLON, XJPX, XPAR, XHKG, XSES.",
    )


class MarketStatusTool(BaseTool):
    """
    LangChain tool that fetches a cryptographically signed market status receipt
    from Headless Oracle V5.

    Returns OPEN, CLOSED, HALTED, or UNKNOWN. Always treat UNKNOWN as CLOSED.
    Verifies the Ed25519 signature before returning the result.
    """

    name: str = "get_market_status"
    description: str = (
        "Prevents trades against closed, halted, or holiday-closed exchanges. "
        "Get the current open/closed status of a stock exchange. "
        "Input: a MIC code (e.g. XNYS for NYSE, XNAS for NASDAQ, XLON for LSE). "
        "Returns the exchange status and whether it is safe to trade. "
        "UNKNOWN must always be treated as CLOSED."
    )
    args_schema: Type[BaseModel] = _MicInput

    api_key: Optional[str] = None
    public_key: Optional[str] = None
    base_url: str = "https://headlessoracle.com"

    def _run(
        self,
        mic: str = "XNYS",
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        with OracleClient(api_key=self.api_key, base_url=self.base_url) as client:
            try:
                receipt = client.get_demo(mic) if not self.api_key else client.get_status(mic)
            except Exception as e:
                # Fail-closed: on any fetch error, report UNKNOWN
                return f"status=UNKNOWN mic={mic} error={e} — treat as CLOSED, halt execution"

        result = verify(receipt, public_key=self.public_key)
        if not result.valid:
            return (
                f"status=UNKNOWN mic={mic} verify_failed={result.reason} "
                f"— signature invalid, treat as CLOSED, halt execution"
            )

        status = receipt["status"]
        safe_to_trade = status == "OPEN"
        reason = receipt.get("reason", "")
        return (
            f"mic={mic} status={status} safe_to_trade={safe_to_trade}"
            + (f" reason={reason}" if reason else "")
            + f" issued_at={receipt['issued_at']} expires_at={receipt['expires_at']}"
        )


class _ScheduleInput(BaseModel):
    mic: str = Field(
        default="XNYS",
        description="Exchange MIC code. Supported: XNYS, XNAS, XLON, XJPX, XPAR, XHKG, XSES.",
    )


class MarketScheduleTool(BaseTool):
    """
    LangChain tool that fetches the next open and close times for a stock exchange
    from Headless Oracle V5.
    """

    name: str = "get_market_schedule"
    description: str = (
        "Get the next open and close times for a stock exchange. "
        "Input: a MIC code (e.g. XNYS for NYSE). "
        "Returns ISO 8601 UTC timestamps for next open and close events."
    )
    args_schema: Type[BaseModel] = _ScheduleInput

    base_url: str = "https://headlessoracle.com"

    def _run(
        self,
        mic: str = "XNYS",
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        with OracleClient(base_url=self.base_url) as client:
            try:
                data = client.get_schedule(mic)
            except Exception as e:
                return f"schedule_unavailable mic={mic} error={e}"

        next_open = data.get("next_open", "unknown")
        next_close = data.get("next_close", "unknown")
        tz = data.get("timezone", "")
        return f"mic={mic} next_open={next_open} next_close={next_close} timezone={tz}"