85 lines
3.9 KiB
Python
85 lines
3.9 KiB
Python
from .base_strategy import BaseStrategy
|
|
from fetchers.factory import FetcherFactory
|
|
from analysis.hk_analyzer import HK_Analyzer
|
|
from reporting.hk_report_generator import HK_ReportGenerator
|
|
from storage.file_io import DataStorage
|
|
import os
|
|
import pandas as pd
|
|
|
|
class HK_Strategy(BaseStrategy):
|
|
def __init__(self, stock_code, ifind_refresh_token):
|
|
super().__init__(stock_code)
|
|
self.refresh_token = ifind_refresh_token
|
|
self.fetcher = FetcherFactory.get_fetcher('HK', ifind_refresh_token=self.refresh_token)
|
|
self.analyzer = HK_Analyzer()
|
|
self.reporter = HK_ReportGenerator()
|
|
self.storage = DataStorage()
|
|
self.raw_data = {}
|
|
self.analysis_result = None
|
|
|
|
def fetch_data(self):
|
|
print(f"Fetching data for HK market, stock: {self.stock_code}")
|
|
# Fetch Financial Statements
|
|
self.raw_data['income'] = self.fetcher.get_income_statement(self.stock_code)
|
|
self.raw_data['balance'] = self.fetcher.get_balance_sheet(self.stock_code)
|
|
self.raw_data['cashflow'] = self.fetcher.get_cash_flow(self.stock_code)
|
|
|
|
# Rename 'end_date' to 'date' for analyzer compatibility
|
|
for key in ['income', 'balance', 'cashflow']:
|
|
if not self.raw_data[key].empty and 'end_date' in self.raw_data[key].columns:
|
|
self.raw_data[key] = self.raw_data[key].rename(columns={'end_date': 'date'})
|
|
|
|
# Fetch Market Metrics
|
|
self.raw_data['metrics'] = self.fetcher.get_market_metrics(self.stock_code)
|
|
|
|
# Fetch Historical Metrics
|
|
dates = []
|
|
if not self.raw_data['income'].empty and 'date' in self.raw_data['income'].columns:
|
|
dates = self.raw_data['income']['date'].tolist()
|
|
|
|
self.raw_data['historical_metrics'] = self.fetcher.get_historical_metrics(self.stock_code, dates)
|
|
|
|
# Fetch Dividends
|
|
self.raw_data['dividends'] = self.fetcher.get_dividends(self.stock_code)
|
|
|
|
# Fetch Repurchases
|
|
self.raw_data['repurchases'] = self.fetcher.get_repurchases(self.stock_code)
|
|
|
|
# Fetch Employee Count
|
|
self.raw_data['employee_count'] = self.fetcher.get_employee_count(self.stock_code)
|
|
|
|
def analyze_data(self):
|
|
print(f"Analyzing data for HK market, stock: {self.stock_code}")
|
|
self.analysis_result = self.analyzer.process_data(
|
|
self.raw_data['income'],
|
|
self.raw_data['balance'],
|
|
self.raw_data['cashflow'],
|
|
self.raw_data['metrics'],
|
|
self.raw_data.get('historical_metrics'),
|
|
self.raw_data.get('dividends'),
|
|
self.raw_data.get('repurchases'),
|
|
self.raw_data.get('employee_count')
|
|
)
|
|
|
|
if self.analysis_result is not None and not self.analysis_result.empty:
|
|
latest = self.analysis_result.iloc[0]
|
|
for m_key, raw_key in [('PE', 'pe'), ('PB', 'pb'), ('DividendYield', 'dividend_yield'), ('Price', 'price'), ('MarketCap', 'market_cap')]:
|
|
if m_key in latest and pd.notna(latest[m_key]) and latest[m_key] != 0:
|
|
self.raw_data['metrics'][raw_key] = latest[m_key]
|
|
|
|
def generate_report(self):
|
|
print(f"Generating report for HK market, stock: {self.stock_code}")
|
|
if self.analysis_result is not None and not self.analysis_result.empty:
|
|
output_dir = os.path.join("data", 'HK', self.stock_code)
|
|
self.reporter.generate_report(
|
|
df_analysis=self.analysis_result,
|
|
symbol=self.stock_code,
|
|
market='HK',
|
|
metrics=self.raw_data['metrics'],
|
|
output_dir=output_dir,
|
|
data_source=getattr(self.fetcher, 'data_source', 'iFinD')
|
|
)
|
|
else:
|
|
print("No analysis result to generate report.")
|
|
raise ValueError(f"Analysis result is empty for {self.stock_code}. Data fetching might have failed.")
|