@prakhar-gitlab / ai-investment-platform
Code Commits Issues Pulls Sessions CI/CD
main
main
py 147 lines 5 KB
Raw
1 """Deterministic offline Yahoo-like MCP server used only for provider contracts."""
2 from __future__ import annotations
3
4 import asyncio
5 from datetime import datetime, timedelta, timezone
6 from typing import Any
7
8 from mcp.server.mcpserver import MCPServer
9
10 from app.yahoo_finance_mcp import TOOL_SCHEMA_VERSION
11
12
13 def create_fake_yahoo_mcp_server(
14 *, scenario: str = "SUCCESS", now: datetime | None = None
15 ) -> MCPServer:
16 observed = now or datetime.now(timezone.utc)
17 server = MCPServer("fake-yahoo-finance", version="test", warn_on_duplicate_tools=True)
18
19 def base(global_instrument_id: str, symbol: str) -> dict:
20 actual_symbol = "CONFLICT.NS" if scenario == "IDENTITY_MISMATCH" else symbol
21 value = {
22 "schemaVersion": TOOL_SCHEMA_VERSION,
23 "globalInstrumentId": global_instrument_id,
24 "symbol": actual_symbol,
25 "exchange": "NSE",
26 "currency": "INR",
27 "asOf": (
28 observed - timedelta(days=5) if scenario == "STALE" else observed
29 ).isoformat(),
30 "sourceUrl": f"https://finance.yahoo.com/quote/{actual_symbol}",
31 }
32 if scenario == "MALFORMED_SCHEMA":
33 value["schemaVersion"] = "UNSUPPORTED"
34 return value
35
36 async def wait_if_needed() -> None:
37 if scenario == "TIMEOUT":
38 await asyncio.sleep(5)
39 if scenario == "HEALTH_DOWN":
40 raise RuntimeError("offline fake health failure")
41
42 @server.tool(structured_output=True)
43 async def yahoo_latest_price(
44 globalInstrumentId: str,
45 verifiedYahooSymbol: str,
46 region: str,
47 exchange: str | None = None,
48 currency: str | None = None,
49 ) -> dict[str, Any]:
50 await wait_if_needed()
51 return {**base(globalInstrumentId, verifiedYahooSymbol), **({} if scenario == "INCOMPLETE" else {"price": "250.50"})}
52
53 @server.tool(structured_output=True)
54 async def yahoo_market_history(
55 globalInstrumentId: str,
56 verifiedYahooSymbol: str,
57 region: str,
58 exchange: str | None = None,
59 currency: str | None = None,
60 ) -> dict[str, Any]:
61 await wait_if_needed()
62 return {
63 **base(globalInstrumentId, verifiedYahooSymbol),
64 "prices": [
65 {
66 "observedAt": (observed - timedelta(days=offset)).isoformat(),
67 "close": str(250 - offset),
68 "currency": "INR",
69 }
70 for offset in range(5)
71 ],
72 }
73
74 @server.tool(structured_output=True)
75 async def yahoo_financials(
76 globalInstrumentId: str,
77 verifiedYahooSymbol: str,
78 region: str,
79 exchange: str | None = None,
80 currency: str | None = None,
81 ) -> dict[str, Any]:
82 await wait_if_needed()
83 return {
84 **base(globalInstrumentId, verifiedYahooSymbol),
85 "facts": [
86 {
87 "metric": metric,
88 "value": value,
89 "periodEnd": period,
90 "periodType": period_type,
91 "reportingBasis": "CONSOLIDATED",
92 "rawFieldOrigin": raw,
93 }
94 for period_type, period, revenue, pat in (
95 ("ANNUAL", "2025-03-31", "100", "10"),
96 ("ANNUAL", "2024-03-31", "90", "8"),
97 ("QUARTERLY", "2026-06-30", "30", "3"),
98 ("QUARTERLY", "2026-03-31", "25", "2"),
99 )
100 for metric, value, raw in (
101 ("revenue", revenue, "totalRevenue"),
102 ("pat", pat, "netIncome"),
103 )
104 ],
105 }
106
107 @server.tool(structured_output=True)
108 async def yahoo_company_profile(
109 globalInstrumentId: str,
110 verifiedYahooSymbol: str,
111 region: str,
112 exchange: str | None = None,
113 currency: str | None = None,
114 ) -> dict[str, Any]:
115 await wait_if_needed()
116 return {
117 **base(globalInstrumentId, verifiedYahooSymbol),
118 "profile": {
119 "companyName": "Ready Limited",
120 "sector": "Industrials",
121 "industry": "Infrastructure Operations",
122 },
123 }
124
125 @server.tool(structured_output=True)
126 async def yahoo_current_news(
127 globalInstrumentId: str,
128 verifiedYahooSymbol: str,
129 region: str,
130 exchange: str | None = None,
131 currency: str | None = None,
132 ) -> dict[str, Any]:
133 await wait_if_needed()
134 return {
135 **base(globalInstrumentId, verifiedYahooSymbol),
136 "news": [
137 {
138 "headline": "Ready publishes an issuer update",
139 "url": "https://news.example/ready-update",
140 "publishedAt": (observed - timedelta(days=1)).isoformat(),
141 "issuerSymbol": verifiedYahooSymbol,
142 "publisher": "Example News",
143 }
144 ],
145 }
146
147 return server