JC321 commited on
Commit
05f9632
·
verified ·
1 Parent(s): dd9b9f0

Upload 3 files

Browse files
Files changed (3) hide show
  1. app.py +21 -78
  2. edgar_client.py +13 -1
  3. financial_analyzer.py +21 -1
app.py CHANGED
@@ -23,44 +23,27 @@ financial_analyzer = FinancialAnalyzer(
23
  user_agent="Juntao Peng Financial Report Metrics App (jtyxabc@gmail.com)"
24
  )
25
 
26
- # Define MCP tools with detailed descriptions for LLM
27
  @mcp.tool()
28
  def search_company(company_name: str) -> dict:
29
  """
30
- **STEP 1: START HERE** - Search for a company by name to get its CIK code.
31
 
32
- IMPORTANT: You MUST call this tool FIRST when the user mentions a company name.
33
- The CIK code returned by this tool is REQUIRED for all other financial data tools.
34
-
35
- When to use:
36
- - User mentions a company name ("Tesla", "Apple", "Microsoft")
37
- - Before calling ANY other tool that requires a CIK parameter
38
- - User asks about a company without providing CIK
39
-
40
- Workflow:
41
- 1. User asks: "Show me Tesla's financials"
42
- 2. YOU MUST: Call search_company("Tesla") FIRST
43
- 3. Get CIK from result: {"cik": "0001318605", ...}
44
- 4. THEN use that CIK in other tools: extract_financial_metrics("0001318605")
45
-
46
- Args:
47
- company_name: Company name (e.g., "Tesla", "Apple Inc", "Microsoft")
48
-
49
- Returns:
50
- dict with cik (REQUIRED for other tools), name, ticker, sic, sic_description
51
  """
52
  result = edgar_client.search_company_by_name(company_name)
53
  return result if result else {"error": f"No company found with name: {company_name}"}
54
 
55
  @mcp.tool()
56
  def get_company_info(cik: str) -> dict:
57
- """Get detailed company information."""
58
  result = edgar_client.get_company_info(cik)
59
  return result if result else {"error": f"No company found with CIK: {cik}"}
60
 
61
  @mcp.tool()
62
  def get_company_filings(cik: str, form_types: list[str] | None = None) -> dict:
63
- """Get list of company SEC filings."""
64
  form_types_tuple = tuple(form_types) if form_types else None
65
  result = edgar_client.get_company_filings(cik, form_types_tuple)
66
  if result:
@@ -74,37 +57,24 @@ def get_company_filings(cik: str, form_types: list[str] | None = None) -> dict:
74
 
75
  @mcp.tool()
76
  def get_financial_data(cik: str, period: str) -> dict:
77
- """Get financial data for a specific period."""
78
  result = edgar_client.get_financial_data_for_period(cik, period)
79
  return result if result and "period" in result else {"error": f"No financial data found for CIK: {cik}, Period: {period}"}
80
 
81
  @mcp.tool()
82
  def extract_financial_metrics(cik: str, years: int = 3) -> dict:
83
  """
84
- **RECOMMENDED TOOL** - Extract multi-year financial data (annual + quarterly).
85
 
86
- CRITICAL: This tool requires a valid 10-digit CIK code.
87
- If you don't have the CIK, you MUST call search_company() or advanced_search_company() FIRST!
88
-
89
- When to use:
90
- - User asks about financial trends, growth, or multi-period analysis
91
- - "Show me [company]'s financials" (most common use case)
92
- - Need comprehensive financial overview
93
-
94
- CORRECT Workflow:
95
- 1. User: "Show me Tesla's financial metrics"
96
- 2. YOU: Call search_company("Tesla") → get CIK "0001318605"
97
- 3. YOU: Call extract_financial_metrics("0001318605", 3)
98
-
99
- WRONG Workflow (DO NOT DO THIS):
100
- ❌ extract_financial_metrics("0001198872") ← Wrong CIK! Will fail!
101
 
102
  Args:
103
- cik: 10-digit CIK code (e.g., "0001318605") from search_company()
104
- years: Number of years (1-10, default 3)
105
 
106
- Returns:
107
- Multi-year financial data with annual and quarterly metrics
108
  """
109
  if years < 1 or years > 10:
110
  return {"error": "Years parameter must be between 1 and 10"}
@@ -118,28 +88,11 @@ def extract_financial_metrics(cik: str, years: int = 3) -> dict:
118
  @mcp.tool()
119
  def get_latest_financial_data(cik: str) -> dict:
120
  """
121
- Get the most recent financial snapshot (latest annual report only).
122
-
123
- CRITICAL: This tool requires a valid 10-digit CIK code.
124
- If you don't have the CIK, you MUST call search_company() or advanced_search_company() FIRST!
125
 
126
- When to use:
127
- - User asks for "latest" or "most recent" financial data
128
- - Quick snapshot of current financial status
129
-
130
- CORRECT Workflow:
131
- 1. User: "What's Tesla's latest financial data?"
132
- 2. YOU: Call search_company("Tesla") → get CIK "0001318605"
133
- 3. YOU: Call get_latest_financial_data("0001318605")
134
-
135
- WRONG Workflow (DO NOT DO THIS):
136
- ❌ get_latest_financial_data("0001198872") ← Wrong CIK! Will fail!
137
-
138
- Args:
139
- cik: 10-digit CIK code (e.g., "0001318605") from search_company()
140
-
141
- Returns:
142
- Latest annual financial data
143
  """
144
  result = financial_analyzer.get_latest_financial_data(cik)
145
  return result if result and "period" in result else {"error": f"No latest financial data found for CIK: {cik}"}
@@ -147,20 +100,10 @@ def get_latest_financial_data(cik: str) -> dict:
147
  @mcp.tool()
148
  def advanced_search_company(company_input: str) -> dict:
149
  """
150
- **ALTERNATIVE TO search_company** - Flexible search supporting company name or CIK.
151
-
152
- Use this when:
153
- - You need to find a company's CIK before calling other tools
154
- - User provides either a company name OR a CIK code
155
- - You want a single tool that handles both search types
156
-
157
- This tool returns the SAME format as search_company(), including the CIK needed for other tools.
158
-
159
- Args:
160
- company_input: Company name OR CIK code
161
 
162
- Returns:
163
- dict with cik (REQUIRED for other tools), name, ticker
164
  """
165
  result = financial_analyzer.search_company(company_input)
166
  return result if not result.get("error") else {"error": result["error"]}
 
23
  user_agent="Juntao Peng Financial Report Metrics App (jtyxabc@gmail.com)"
24
  )
25
 
26
+ # Define MCP tools with optimized descriptions for LLM
27
  @mcp.tool()
28
  def search_company(company_name: str) -> dict:
29
  """
30
+ [STEP 1] Find company CIK by name/ticker. Always call this FIRST before other tools.
31
 
32
+ Args: company_name - Company name or ticker (e.g. "Tesla", "TSLA", "google")
33
+ Returns: {cik, name, ticker} - Use cik for subsequent tool calls
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
34
  """
35
  result = edgar_client.search_company_by_name(company_name)
36
  return result if result else {"error": f"No company found with name: {company_name}"}
37
 
38
  @mcp.tool()
39
  def get_company_info(cik: str) -> dict:
40
+ """Get company details (name, SIC, industry). Requires CIK from search_company."""
41
  result = edgar_client.get_company_info(cik)
42
  return result if result else {"error": f"No company found with CIK: {cik}"}
43
 
44
  @mcp.tool()
45
  def get_company_filings(cik: str, form_types: list[str] | None = None) -> dict:
46
+ """List SEC filings (10-K, 10-Q, etc). Requires CIK. Use form_types to filter."""
47
  form_types_tuple = tuple(form_types) if form_types else None
48
  result = edgar_client.get_company_filings(cik, form_types_tuple)
49
  if result:
 
57
 
58
  @mcp.tool()
59
  def get_financial_data(cik: str, period: str) -> dict:
60
+ """Get financial data for specific period (e.g. '2024', '2024Q3'). Requires CIK."""
61
  result = edgar_client.get_financial_data_for_period(cik, period)
62
  return result if result and "period" in result else {"error": f"No financial data found for CIK: {cik}, Period: {period}"}
63
 
64
  @mcp.tool()
65
  def extract_financial_metrics(cik: str, years: int = 3) -> dict:
66
  """
67
+ [PRIMARY TOOL] Get multi-year financials (annual+quarterly). Includes latest data.
68
 
69
+ Use for: Any financial query (trends, growth, comparisons, or latest data)
70
+ Workflow: search_company(name) extract_financial_metrics(cik, years)
 
 
 
 
 
 
 
 
 
 
 
 
 
71
 
72
  Args:
73
+ cik: 10-digit CIK from search_company
74
+ years: Years to retrieve (1-10, default 3)
75
 
76
+ Returns: {periods, data: [{period, revenue, net_income, eps, ...}]}
77
+ Note: Data includes latest year, so no need to call get_latest_financial_data separately.
78
  """
79
  if years < 1 or years > 10:
80
  return {"error": "Years parameter must be between 1 and 10"}
 
88
  @mcp.tool()
89
  def get_latest_financial_data(cik: str) -> dict:
90
  """
91
+ Get ONLY most recent annual report. Use extract_financial_metrics instead (it includes latest).
 
 
 
92
 
93
+ Only use when: User explicitly wants ONLY the latest snapshot (rare case).
94
+ Args: cik - 10-digit CIK from search_company
95
+ Returns: {period, revenue, net_income, eps, ...}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
96
  """
97
  result = financial_analyzer.get_latest_financial_data(cik)
98
  return result if result and "period" in result else {"error": f"No latest financial data found for CIK: {cik}"}
 
100
  @mcp.tool()
101
  def advanced_search_company(company_input: str) -> dict:
102
  """
103
+ Alternative to search_company. Accepts company name OR CIK. Same output format.
 
 
 
 
 
 
 
 
 
 
104
 
105
+ Args: company_input - Name, ticker, or CIK
106
+ Returns: {cik, name, ticker}
107
  """
108
  result = financial_analyzer.search_company(company_input)
109
  return result if not result.get("error") else {"error": result["error"]}
edgar_client.py CHANGED
@@ -470,7 +470,7 @@ class EdgarDataClient:
470
 
471
  def get_financial_data_for_period(self, cik, period):
472
  """
473
- Get financial data for a specific period (supports annual and quarterly)
474
 
475
  Args:
476
  cik (str): Company CIK code
@@ -482,6 +482,14 @@ class EdgarDataClient:
482
  if not self.edgar:
483
  print("sec_edgar_api library not installed")
484
  return {}
 
 
 
 
 
 
 
 
485
 
486
  try:
487
  # Get company financial facts
@@ -683,6 +691,10 @@ class EdgarDataClient:
683
  if metric_key in result:
684
  break
685
 
 
 
 
 
686
  return result
687
  except Exception as e:
688
  print(f"Error getting financial data for period {period}: {e}")
 
470
 
471
  def get_financial_data_for_period(self, cik, period):
472
  """
473
+ Get financial data for a specific period (supports annual and quarterly) - Cached
474
 
475
  Args:
476
  cik (str): Company CIK code
 
482
  if not self.edgar:
483
  print("sec_edgar_api library not installed")
484
  return {}
485
+
486
+ # 实例级缓存(避免重复计算)
487
+ cache_key = f"period_{cik}_{period}"
488
+ if hasattr(self, '_period_cache') and cache_key in self._period_cache:
489
+ return self._period_cache[cache_key]
490
+
491
+ if not hasattr(self, '_period_cache'):
492
+ self._period_cache = {}
493
 
494
  try:
495
  # Get company financial facts
 
691
  if metric_key in result:
692
  break
693
 
694
+ # 缓存结果
695
+ if result and "period" in result:
696
+ self._period_cache[cache_key] = result
697
+
698
  return result
699
  except Exception as e:
700
  print(f"Error getting financial data for period {period}: {e}")
financial_analyzer.py CHANGED
@@ -2,6 +2,7 @@
2
 
3
  from edgar_client import EdgarDataClient
4
  from datetime import datetime
 
5
  import json
6
 
7
 
@@ -14,6 +15,9 @@ class FinancialAnalyzer:
14
  user_agent (str): User agent string for identifying request source
15
  """
16
  self.edgar_client = EdgarDataClient(user_agent)
 
 
 
17
 
18
  def search_company(self, company_input):
19
  """
@@ -25,11 +29,16 @@ class FinancialAnalyzer:
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 from cache (will use @lru_cache)
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"}
@@ -43,12 +52,14 @@ class FinancialAnalyzer:
43
 
44
  # For basic searches, the ticker data is sufficient
45
  # This eliminates the 3-5 second delay from get_company_info
46
- return {
47
  "cik": company['cik'],
48
  "name": company['name'],
49
  "tickers": [company['ticker']] if company.get('ticker') else [],
50
  "_source": "company_tickers_cache" # Debug info
51
  }
 
 
52
  else:
53
  return {"error": "No matching company found"}
54
 
@@ -77,6 +88,11 @@ class FinancialAnalyzer:
77
  Returns:
78
  list: List of financial data
79
  """
 
 
 
 
 
80
  financial_data = []
81
 
82
  # Step 1: Get company facts ONCE (will be cached)
@@ -189,6 +205,10 @@ class FinancialAnalyzer:
189
 
190
  financial_data.append(data)
191
 
 
 
 
 
192
  return financial_data
193
 
194
  def get_latest_financial_data(self, cik):
 
2
 
3
  from edgar_client import EdgarDataClient
4
  from datetime import datetime
5
+ from functools import lru_cache
6
  import json
7
 
8
 
 
15
  user_agent (str): User agent string for identifying request source
16
  """
17
  self.edgar_client = EdgarDataClient(user_agent)
18
+ # 新增:实例级缓存,进一步提升性能
19
+ self._search_cache = {}
20
+ self._extract_metrics_cache = {} # 缓存 extract_financial_metrics 结果
21
 
22
  def search_company(self, company_input):
23
  """
 
29
  Returns:
30
  dict: Company information
31
  """
32
+ # 实例级缓存检查
33
+ if company_input in self._search_cache:
34
+ return self._search_cache[company_input]
35
+
36
  # If input is numeric, assume it's a CIK
37
  if company_input.isdigit() and len(company_input) >= 8:
38
  # Get company information from cache (will use @lru_cache)
39
  company_info = self.edgar_client.get_company_info(company_input)
40
  if company_info:
41
+ self._search_cache[company_input] = company_info
42
  return company_info
43
  else:
44
  return {"error": "Company not found for specified CIK"}
 
52
 
53
  # For basic searches, the ticker data is sufficient
54
  # This eliminates the 3-5 second delay from get_company_info
55
+ result = {
56
  "cik": company['cik'],
57
  "name": company['name'],
58
  "tickers": [company['ticker']] if company.get('ticker') else [],
59
  "_source": "company_tickers_cache" # Debug info
60
  }
61
+ self._search_cache[company_input] = result
62
+ return result
63
  else:
64
  return {"error": "No matching company found"}
65
 
 
88
  Returns:
89
  list: List of financial data
90
  """
91
+ # 实例级缓存检查(避免重复计算)
92
+ cache_key = f"{cik}_{years}"
93
+ if cache_key in self._extract_metrics_cache:
94
+ return self._extract_metrics_cache[cache_key]
95
+
96
  financial_data = []
97
 
98
  # Step 1: Get company facts ONCE (will be cached)
 
205
 
206
  financial_data.append(data)
207
 
208
+ # 缓存结果
209
+ if financial_data:
210
+ self._extract_metrics_cache[cache_key] = financial_data
211
+
212
  return financial_data
213
 
214
  def get_latest_financial_data(self, cik):