JC321 commited on
Commit
0c8197e
·
verified ·
1 Parent(s): 97d4ed3

Upload 3 files

Browse files
Files changed (3) hide show
  1. edgar_client.py +34 -16
  2. financial_analyzer.py +54 -23
  3. mcp_server_sse.py +3 -3
edgar_client.py CHANGED
@@ -118,42 +118,60 @@ class EdgarDataClient:
118
  return None
119
 
120
  def search_company_by_name(self, company_name):
121
- """Search company CIK by company name"""
122
  try:
123
- # Use SEC company ticker database
124
- url = "https://www.sec.gov/files/company_tickers.json"
 
125
 
126
- response = self._make_request_with_retry(url)
127
- if not response:
128
- return None
129
-
130
- companies = response.json()
 
 
 
 
 
 
131
 
132
  # Search for matching company names
133
  matches = []
134
  exact_matches = []
 
 
 
135
  for _, company in companies.items():
136
  company_title = company["title"].lower()
137
- search_name = company_name.lower()
138
 
139
- # Exact match
140
- if search_name == company_title:
 
 
 
 
 
 
 
141
  exact_matches.append({
142
  "cik": str(company["cik_str"]).zfill(10),
143
  "name": company["title"],
144
  "ticker": company["ticker"]
145
  })
146
- # Partial match
147
- elif search_name in company_title or \
148
- search_name in company["ticker"].lower():
149
  matches.append({
150
  "cik": str(company["cik_str"]).zfill(10),
151
  "name": company["title"],
152
  "ticker": company["ticker"]
153
  })
154
 
155
- # Return exact match first, then partial match
156
- if exact_matches:
 
 
157
  return exact_matches[0]
158
  elif matches:
159
  return matches[0]
 
118
  return None
119
 
120
  def search_company_by_name(self, company_name):
121
+ """Search company CIK by company name with caching"""
122
  try:
123
+ # Check cache for company_tickers.json
124
+ cache_key = "company_tickers_json"
125
+ companies = self._get_cached(cache_key)
126
 
127
+ if not companies:
128
+ # Use SEC company ticker database
129
+ url = "https://www.sec.gov/files/company_tickers.json"
130
+
131
+ response = self._make_request_with_retry(url)
132
+ if not response:
133
+ return None
134
+
135
+ companies = response.json()
136
+ # Cache the entire company list for 5 minutes
137
+ self._set_cache(cache_key, companies)
138
 
139
  # Search for matching company names
140
  matches = []
141
  exact_matches = []
142
+ ticker_matches = []
143
+ search_name = company_name.lower().strip()
144
+
145
  for _, company in companies.items():
146
  company_title = company["title"].lower()
147
+ company_ticker = company["ticker"].lower()
148
 
149
+ # Exact ticker match (highest priority)
150
+ if search_name == company_ticker:
151
+ ticker_matches.append({
152
+ "cik": str(company["cik_str"]).zfill(10),
153
+ "name": company["title"],
154
+ "ticker": company["ticker"]
155
+ })
156
+ # Exact name match
157
+ elif search_name == company_title:
158
  exact_matches.append({
159
  "cik": str(company["cik_str"]).zfill(10),
160
  "name": company["title"],
161
  "ticker": company["ticker"]
162
  })
163
+ # Partial match in name or ticker
164
+ elif search_name in company_title or search_name in company_ticker:
 
165
  matches.append({
166
  "cik": str(company["cik_str"]).zfill(10),
167
  "name": company["title"],
168
  "ticker": company["ticker"]
169
  })
170
 
171
+ # Return ticker match first, then exact name match, then partial match
172
+ if ticker_matches:
173
+ return ticker_matches[0]
174
+ elif exact_matches:
175
  return exact_matches[0]
176
  elif matches:
177
  return matches[0]
financial_analyzer.py CHANGED
@@ -17,51 +17,82 @@ class FinancialAnalyzer:
17
 
18
  def search_company(self, company_input):
19
  """
20
- Search company information (by name or CIK)
21
 
22
  Args:
23
- company_input (str): Company name or CIK
24
 
25
  Returns:
26
  dict: Company information
27
  """
28
- # If input is numeric, assume it's a CIK
 
 
 
29
  if company_input.isdigit() and len(company_input) >= 8:
30
- # Get company information
31
- company_info = self.edgar_client.get_company_info(company_input)
 
32
  if company_info:
33
  return company_info
34
  else:
35
  return {"error": "Company not found for specified CIK"}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
36
  else:
37
- # Search company by name
38
- company = self.edgar_client.search_company_by_name(company_input)
39
- if company:
40
- # Get detailed information
41
- company_info = self.edgar_client.get_company_info(company['cik'])
42
- if company_info:
43
- return company_info
44
- else:
45
- # If detailed info unavailable, return basic info
46
- return {
47
- "cik": company['cik'],
48
- "name": company['name'],
49
- "tickers": [company['ticker']] if company['ticker'] else []
50
- }
51
- else:
52
- return {"error": "No matching company found"}
53
 
54
- def get_company_filings_list(self, cik, form_types=['10-K', '10-Q']):
55
  """
56
  Get company filings list
57
 
58
  Args:
59
  cik (str): Company CIK
60
- form_types (list): List of form types
61
 
62
  Returns:
63
  list: Filings list
64
  """
 
 
65
  filings = self.edgar_client.get_company_filings(cik, form_types)
66
  return filings
67
 
 
17
 
18
  def search_company(self, company_input):
19
  """
20
+ Search company information (by name, ticker, or CIK) - Optimized version
21
 
22
  Args:
23
+ company_input (str): Company name, ticker, or CIK
24
 
25
  Returns:
26
  dict: Company information
27
  """
28
+ # Strip whitespace
29
+ company_input = company_input.strip()
30
+
31
+ # Strategy 1: If input is numeric and looks like CIK (8-10 digits), treat as CIK
32
  if company_input.isdigit() and len(company_input) >= 8:
33
+ # Pad to 10 digits if needed
34
+ cik = company_input.zfill(10)
35
+ company_info = self.edgar_client.get_company_info(cik)
36
  if company_info:
37
  return company_info
38
  else:
39
  return {"error": "Company not found for specified CIK"}
40
+
41
+ # Strategy 2: Search by name/ticker
42
+ # This returns basic info: {cik, name, ticker}
43
+ basic_info = self.edgar_client.search_company_by_name(company_input)
44
+
45
+ if not basic_info:
46
+ return {"error": "No matching company found"}
47
+
48
+ # Strategy 3: Decide whether to fetch detailed info
49
+ # If user input is short (likely a ticker), return enriched basic info quickly
50
+ # If user input is long (likely a full name), get detailed info
51
+
52
+ input_length = len(company_input)
53
+ is_likely_ticker = input_length <= 5 and company_input.isupper()
54
+
55
+ # For ticker searches, return enriched basic info without additional API call
56
+ if is_likely_ticker:
57
+ # Quick response with basic info + some enrichment
58
+ return {
59
+ "cik": basic_info['cik'],
60
+ "name": basic_info['name'],
61
+ "tickers": [basic_info['ticker']] if basic_info.get('ticker') else [],
62
+ "ein": None, # Not available in basic search
63
+ "fiscal_year_end": None, # Not available in basic search
64
+ "sic_description": None, # Not available in basic search
65
+ "_source": "quick_search",
66
+ "_note": "Basic info from ticker search. Use get_company_info for full details."
67
+ }
68
+
69
+ # For name searches, fetch detailed info (worth the extra API call)
70
+ company_info = self.edgar_client.get_company_info(basic_info['cik'])
71
+
72
+ if company_info:
73
+ return company_info
74
  else:
75
+ # Fallback to basic info if detailed fetch fails
76
+ return {
77
+ "cik": basic_info['cik'],
78
+ "name": basic_info['name'],
79
+ "tickers": [basic_info['ticker']] if basic_info.get('ticker') else [],
80
+ "_source": "basic_search_fallback"
81
+ }
 
 
 
 
 
 
 
 
 
82
 
83
+ def get_company_filings_list(self, cik, form_types=None):
84
  """
85
  Get company filings list
86
 
87
  Args:
88
  cik (str): Company CIK
89
+ form_types (list): List of form types (default: ['10-K', '10-Q'])
90
 
91
  Returns:
92
  list: Filings list
93
  """
94
+ if form_types is None:
95
+ form_types = ['10-K', '10-Q']
96
  filings = self.edgar_client.get_company_filings(cik, form_types)
97
  return filings
98
 
mcp_server_sse.py CHANGED
@@ -24,7 +24,7 @@ from contextlib import contextmanager
24
  app = FastAPI(
25
  title="SEC Financial Report MCP Server",
26
  description="Model Context Protocol Server for SEC EDGAR Financial Data",
27
- version="2.2.3"
28
  )
29
 
30
  # Server startup time for monitoring
@@ -403,7 +403,7 @@ async def handle_mcp_message(request: MCPRequest):
403
  },
404
  "serverInfo": {
405
  "name": "sec-financial-data",
406
- "version": "2.2.3"
407
  }
408
  }
409
  ).dict()
@@ -1280,7 +1280,7 @@ async def health_check():
1280
  return {
1281
  "status": "healthy",
1282
  "server": "sec-financial-data",
1283
- "version": "2.2.3",
1284
  "protocol": "MCP",
1285
  "transport": "SSE",
1286
  "tools_count": len(MCP_TOOLS),
 
24
  app = FastAPI(
25
  title="SEC Financial Report MCP Server",
26
  description="Model Context Protocol Server for SEC EDGAR Financial Data",
27
+ version="2.3.0"
28
  )
29
 
30
  # Server startup time for monitoring
 
403
  },
404
  "serverInfo": {
405
  "name": "sec-financial-data",
406
+ "version": "2.3.0"
407
  }
408
  }
409
  ).dict()
 
1280
  return {
1281
  "status": "healthy",
1282
  "server": "sec-financial-data",
1283
+ "version": "2.3.0",
1284
  "protocol": "MCP",
1285
  "transport": "SSE",
1286
  "tools_count": len(MCP_TOOLS),