99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
import os
|
|
import sys
|
|
from dotenv import load_dotenv
|
|
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), 'src'))
|
|
|
|
from fetchers.factory import FetcherFactory
|
|
from storage.file_io import DataStorage
|
|
from analysis.calculator import FinancialCalculator
|
|
from reporting.markdown_generator import MarkdownReporter
|
|
from reporting.html_generator import HtmlReporter
|
|
|
|
def run_analysis(market, symbol):
|
|
load_dotenv()
|
|
tushare_token = os.getenv('TUSHARE_TOKEN')
|
|
av_key = os.getenv('ALPHA_VANTAGE_KEY')
|
|
|
|
print(f"\nProcessing {market} stock: {symbol}...")
|
|
# 1. Fetch Data
|
|
try:
|
|
fetcher = FetcherFactory.get_fetcher(market, tushare_token, av_key)
|
|
|
|
print("Fetching Income Statement...")
|
|
df_inc = fetcher.get_income_statement(symbol)
|
|
print("Fetching Balance Sheet...")
|
|
df_bal = fetcher.get_balance_sheet(symbol)
|
|
print("Fetching Cash Flow...")
|
|
df_cf = fetcher.get_cash_flow(symbol)
|
|
print("Fetching Market Metrics...")
|
|
metrics = fetcher.get_market_metrics(symbol)
|
|
print(f"Metrics: {metrics}")
|
|
|
|
# 1.1 Fetch Historical Metrics
|
|
historical_metrics_df = None
|
|
if not df_inc.empty:
|
|
# Tushare dates are often YYYYMMDD. Align format.
|
|
dates = df_inc['date'].tolist() if 'date' in df_inc.columns else []
|
|
# Also combine from bal/cf if needed? usually inc covers reporting periods.
|
|
if dates and hasattr(fetcher, 'get_historical_metrics'):
|
|
print(f"Fetching Historical Market Metrics for {len(dates)} periods...")
|
|
historical_metrics_df = fetcher.get_historical_metrics(symbol, dates)
|
|
|
|
# 2. Save Data
|
|
storage = DataStorage()
|
|
storage.save_data(df_inc, market, symbol, "income_statement")
|
|
storage.save_data(df_bal, market, symbol, "balance_sheet")
|
|
storage.save_data(df_cf, market, symbol, "cash_flow")
|
|
|
|
# 3. Calculate Indicators
|
|
print("Calculating Financial Indicators...")
|
|
calculator = FinancialCalculator()
|
|
df_analysis = calculator.process_data(df_inc, df_bal, df_cf, metrics, market, historical_metrics_df)
|
|
|
|
if df_analysis.empty:
|
|
print("Not enough data to calculate indicators.")
|
|
return
|
|
|
|
# 4. Generate Reports
|
|
print("Generating Reports...")
|
|
md_reporter = MarkdownReporter()
|
|
html_reporter = HtmlReporter()
|
|
|
|
md_content = md_reporter.generate_report(df_analysis, market, symbol, metrics)
|
|
html_content = html_reporter.generate_report(df_analysis, market, symbol, metrics)
|
|
|
|
output_dir = os.path.join("data", market, symbol)
|
|
md_path = os.path.join(output_dir, "report.md")
|
|
html_path = os.path.join(output_dir, "report.html")
|
|
|
|
with open(md_path, "w", encoding='utf-8') as f:
|
|
f.write(md_content)
|
|
|
|
with open(html_path, "w", encoding='utf-8') as f:
|
|
f.write(html_content)
|
|
|
|
print(f"\nAnalysis completed! Reports saved to:")
|
|
print(f" - Markdown: {md_path}")
|
|
print(f" - HTML: {html_path}")
|
|
print("-" * 50)
|
|
|
|
except Exception as e:
|
|
print(f"Error occurred: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) > 2:
|
|
market = sys.argv[1]
|
|
symbol = sys.argv[2]
|
|
run_analysis(market, symbol)
|
|
else:
|
|
print("Usage: python main.py <MARKET> <SYMBOL>")
|
|
# Default test cases if run without args (for manual testing)
|
|
print("Running default test cases:")
|
|
# Test CN
|
|
run_analysis('CN', '600519.SH')
|
|
# Test US
|
|
run_analysis('US', 'AAPL')
|