Spaces:
Runtime error
Runtime error
| """Data transformation helpers: clean DataFrames, compute metrics.""" | |
| import pandas as pd | |
| import numpy as np | |
| from typing import List, Dict, Any, Optional | |
| from datetime import date | |
| import structlog | |
| from .parsers import extract_list_from_response, detect_field_names | |
| logger = structlog.get_logger() | |
| def prepare_live_df(stocks: List[Dict]) -> pd.DataFrame: | |
| """Convert live stock list to a clean DataFrame with proper numeric types.""" | |
| if not stocks: | |
| return pd.DataFrame() | |
| df = pd.DataFrame(stocks) | |
| numeric_cols = ['lastTradedPrice', 'previousClose', 'totalTradeQuantity', | |
| 'highPrice', 'lowPrice', 'openPrice'] | |
| for col in numeric_cols: | |
| if col in df.columns: | |
| df[col] = pd.to_numeric(df[col], errors='coerce') | |
| return df | |
| def compute_daily_market_summary(live_raw: Any) -> Dict: | |
| """ | |
| From live raw data, compute total turnover, volume, advancers/decliners. | |
| """ | |
| stocks = extract_list_from_response(live_raw) | |
| if not stocks: | |
| logger.warning("No stock list found in live data") | |
| return {} | |
| sample = stocks[0] | |
| fields = detect_field_names(sample) | |
| if None in fields.values(): | |
| logger.error(f"Missing required fields: {fields}") | |
| return {} | |
| price_key = fields['price_key'] | |
| prev_close_key = fields['prev_close_key'] | |
| volume_key = fields['volume_key'] | |
| df = prepare_live_df(stocks) | |
| df.dropna(subset=[price_key, prev_close_key, volume_key], inplace=True) | |
| df['turnover'] = df[price_key] * df[volume_key] | |
| df['pct_change'] = ((df[price_key] - df[prev_close_key]) / df[prev_close_key]) * 100 | |
| total_turnover = df['turnover'].sum() | |
| total_volume = df[volume_key].sum() | |
| advancers = (df['pct_change'] > 0).sum() | |
| decliners = (df['pct_change'] < 0).sum() | |
| unchanged = (df['pct_change'] == 0).sum() | |
| avg_pct_change = df['pct_change'].mean() | |
| summary = { | |
| "date": date.today().isoformat(), | |
| "total_turnover": float(total_turnover), | |
| "total_volume": int(total_volume), | |
| "advancers": int(advancers), | |
| "decliners": int(decliners), | |
| "unchanged": int(unchanged), | |
| "avg_pct_change": float(avg_pct_change) if pd.notna(avg_pct_change) else 0.0, | |
| "total_companies": len(df) | |
| } | |
| return summary | |
| def compute_daily_market_cap(live_raw: Any, company_master: List[Dict]) -> Dict: | |
| """Compute total market cap and sector breakdown.""" | |
| stocks = extract_list_from_response(live_raw) | |
| if not stocks or not company_master: | |
| logger.warning("Missing data for market cap computation") | |
| return {} | |
| sample = stocks[0] | |
| fields = detect_field_names(sample) | |
| if not fields['price_key'] or not fields['symbol_key']: | |
| logger.error("Price or symbol key not found") | |
| return {} | |
| price_key = fields['price_key'] | |
| symbol_key = fields['symbol_key'] | |
| live_df = pd.DataFrame(stocks) | |
| master_df = pd.DataFrame(company_master) | |
| if 'symbol' not in master_df.columns: | |
| logger.error("Master data missing 'symbol' column") | |
| return {} | |
| live_df[price_key] = pd.to_numeric(live_df[price_key], errors='coerce') | |
| live_df = live_df[[symbol_key, price_key]].dropna() | |
| merged = pd.merge( | |
| live_df, | |
| master_df[['symbol', 'shares', 'sector_name']], | |
| left_on=symbol_key, | |
| right_on='symbol', | |
| how='inner' | |
| ) | |
| merged['shares'] = merged['shares'].astype(str).str.replace(',', '').str.strip() | |
| merged['shares'] = pd.to_numeric(merged['shares'], errors='coerce') | |
| merged['market_cap'] = merged[price_key] * merged['shares'] | |
| total_market_cap = merged['market_cap'].sum() | |
| sector_caps = merged.groupby('sector_name')['market_cap'].sum().to_dict() | |
| result = { | |
| "date": date.today().isoformat(), | |
| "total_market_cap": float(total_market_cap), | |
| "sector_breakdown": sector_caps, | |
| "count_companies_used": len(merged) | |
| } | |
| return result | |
| def compute_price_volume_analysis(live_raw: Any) -> List[Dict]: | |
| """Generate flags for high volume and big movers.""" | |
| stocks = extract_list_from_response(live_raw) | |
| if not stocks: | |
| return [] | |
| sample = stocks[0] | |
| fields = detect_field_names(sample) | |
| if None in fields.values(): | |
| logger.error("Missing required fields for price-volume analysis") | |
| return [] | |
| price_key = fields['price_key'] | |
| prev_close_key = fields['prev_close_key'] | |
| volume_key = fields['volume_key'] | |
| symbol_key = fields['symbol_key'] | |
| df = prepare_live_df(stocks) | |
| df.dropna(subset=[price_key, prev_close_key, volume_key], inplace=True) | |
| df['turnover'] = df[price_key] * df[volume_key] | |
| df['pct_change'] = ((df[price_key] - df[prev_close_key]) / df[prev_close_key]) * 100 | |
| median_volume = df[volume_key].median() | |
| df['high_volume'] = df[volume_key] > median_volume * 2 | |
| df['big_mover'] = abs(df['pct_change']) > 5 | |
| results = df[[ | |
| symbol_key, price_key, volume_key, 'turnover', 'pct_change', | |
| 'high_volume', 'big_mover' | |
| ]].to_dict(orient='records') | |
| return results |