Spaces:
Running
Running
File size: 2,432 Bytes
70307c7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """Sector peer return features for ML training (C30).
Adds prior-day mean return of same-sector peers as non-leaking features:
peer_ret_1d — yesterday's 1-day return of sector peers (mean)
peer_ret_5d — 5-day return of peers ending yesterday (mean)
Stocks with no peers (telecom) get 0.0 fill.
"""
from __future__ import annotations
import logging
from typing import Sequence
import numpy as np
import pandas as pd
logger = logging.getLogger(__name__)
SECTOR_MAP: dict[str, list[str]] = {
"semis": ["2330", "2454", "2303"],
"electronics": ["2317", "2382", "2308"],
"financials": ["2881", "2882", "2886"],
"etfs": ["0050", "0056"],
"telecom": ["2412"],
}
_STOCK_SECTOR = {s: sec for sec, members in SECTOR_MAP.items() for s in members}
def _peer_list(stock_no: str) -> list[str]:
sector = _STOCK_SECTOR.get(stock_no, "")
return [p for p in SECTOR_MAP.get(sector, []) if p != stock_no]
def add_peer_returns(df: pd.DataFrame, stock_no: str) -> pd.DataFrame:
"""Fetch peer close prices and add peer_ret_1d / peer_ret_5d to df."""
peers = _peer_list(stock_no)
if not peers or "date" not in df.columns:
df["peer_ret_1d"] = 0.0
df["peer_ret_5d"] = 0.0
return df
from services.predictor_service import _fetch_with_cache
date_col = df["date"].astype(str)
r1_series, r5_series = [], []
for peer in peers:
try:
peer_df = _fetch_with_cache(peer, months=24)
if peer_df is None or peer_df.empty or "date" not in peer_df.columns:
continue
close = peer_df.set_index("date")["close"].astype(float)
r1 = close.pct_change(1).shift(1)
r5 = close.pct_change(5).shift(1)
r1_series.append(date_col.map(r1.to_dict()).astype(float))
r5_series.append(date_col.map(r5.to_dict()).astype(float))
except Exception as exc:
logger.debug("peer_returns fetch failed for %s peer %s: %s", stock_no, peer, exc)
if not r1_series:
df["peer_ret_1d"] = 0.0
df["peer_ret_5d"] = 0.0
return df
idx = df.index
df["peer_ret_1d"] = (
pd.concat(r1_series, axis=1).mean(axis=1)
.ffill().bfill().fillna(0.0)
.values
)
df["peer_ret_5d"] = (
pd.concat(r5_series, axis=1).mean(axis=1)
.ffill().bfill().fillna(0.0)
.values
)
return df
|