""" Common utilities for quantitative investment strategies. """ import pandas as pd from db.common import DBManager def get_sector_data(): """Get sector data for all stocks.""" db = DBManager() return db.get_sector_list() def get_stock_data(): """Get stock data for all available stocks.""" db = DBManager() return db.get_ticker_list() def get_financial_data(): """Get financial statement data.""" db = DBManager() return db.get_fs_list() def get_expanded_financial_data(): """Get expanded financial statement data.""" db = DBManager() return db.get_expanded_fs_list() def get_price_data(interval_months=60): """Get price data for the specified interval.""" db = DBManager() return db.get_price_list(interval_months) def get_value_metrics(): """Get value metrics for all stocks.""" db = DBManager() return db.get_value_list() def calculate_zscore(series): """Calculate z-score for a pandas series.""" return (series - series.mean()) / series.std() def rank_by_metric(df, metric, ascending=True): """Rank stocks by a metric.""" return df.sort_values(by=metric, ascending=ascending) def rank_combined(df, metrics, weights=None, ascending=True): """Rank stocks by combined metrics. Args: df: DataFrame with stock data metrics: List of column names to use for ranking weights: Optional weights for each metric ascending: Direction for ranking (True = smaller is better) Returns: DataFrame with combined rank """ if weights is None: weights = [1] * len(metrics) # Calculate rank for each metric ranks = pd.DataFrame(index=df.index) for i, metric in enumerate(metrics): ranks[f'rank_{metric}'] = df[metric].rank(ascending=ascending) * weights[i] # Calculate combined rank ranks['combined_rank'] = ranks.mean(axis=1) # Combine with original data result = pd.concat([df, ranks['combined_rank']], axis=1) return result.sort_values('combined_rank')