You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
"""
|
|
Index Constituents Service
|
|
S&P 500 / Nasdaq 100 구성 종목을 Wikipedia에서 조회
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from io import StringIO
|
|
from typing import Optional
|
|
from urllib.request import Request, urlopen
|
|
|
|
import pandas as pd
|
|
|
|
logger = logging.getLogger("app.services.index_constituents")
|
|
|
|
|
|
class IndexConstituentsService:
|
|
INDEXES = {
|
|
"sp500": {
|
|
"url": "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies",
|
|
"table_index": 0,
|
|
"ticker_col": "Symbol",
|
|
"name_col": "Security",
|
|
"sector_col": "GICS Sector",
|
|
"industry_col": "GICS Sub-Industry",
|
|
},
|
|
"nasdaq100": {
|
|
"url": "https://en.wikipedia.org/wiki/Nasdaq-100",
|
|
"table_index": 4,
|
|
"ticker_col": "Ticker",
|
|
"name_col": "Company",
|
|
"sector_col": "GICS Sector",
|
|
"industry_col": "GICS Sub-Industry",
|
|
},
|
|
}
|
|
|
|
def _fetch_constituents_sync(self, index_name: str) -> dict:
|
|
config = self.INDEXES[index_name]
|
|
url = config["url"]
|
|
ticker_col = config["ticker_col"]
|
|
name_col = config["name_col"]
|
|
sector_col = config["sector_col"]
|
|
industry_col = config["industry_col"]
|
|
|
|
logger.info(f"Fetching Wikipedia tables from {url}")
|
|
req = Request(url, headers={"User-Agent": "Mozilla/5.0 (compatible; StockOracle/1.0)"})
|
|
with urlopen(req, timeout=25) as resp:
|
|
html = resp.read().decode("utf-8")
|
|
tables = pd.read_html(StringIO(html), flavor="lxml")
|
|
|
|
# Try configured table index first
|
|
df = None
|
|
table_index = config["table_index"]
|
|
if table_index < len(tables) and ticker_col in tables[table_index].columns:
|
|
df = tables[table_index]
|
|
logger.info(f"Using configured table index {table_index}")
|
|
else:
|
|
# Fallback: scan all tables for ticker column
|
|
for i, tbl in enumerate(tables):
|
|
if ticker_col in tbl.columns:
|
|
df = tbl
|
|
logger.info(f"Fallback: found ticker column '{ticker_col}' in table {i}")
|
|
break
|
|
|
|
if df is None:
|
|
raise ValueError(
|
|
f"Could not find table with column '{ticker_col}' in any of the {len(tables)} tables"
|
|
)
|
|
|
|
constituents = []
|
|
for _, row in df.iterrows():
|
|
symbol = str(row.get(ticker_col, "")).strip()
|
|
if not symbol or symbol == "nan":
|
|
continue
|
|
constituents.append({
|
|
"symbol": symbol,
|
|
"name": str(row.get(name_col, "")).strip(),
|
|
"sector": str(row.get(sector_col, "")).strip() if sector_col in df.columns else None,
|
|
"industry": str(row.get(industry_col, "")).strip() if industry_col in df.columns else None,
|
|
})
|
|
|
|
return {
|
|
"success": True,
|
|
"index": index_name,
|
|
"count": len(constituents),
|
|
"constituents": constituents,
|
|
}
|
|
|
|
async def get_constituents(self, index_name: str) -> dict:
|
|
if index_name not in self.INDEXES:
|
|
raise ValueError(f"Unknown index '{index_name}'. Supported: {list(self.INDEXES.keys())}")
|
|
|
|
loop = asyncio.get_event_loop()
|
|
result = await loop.run_in_executor(None, self._fetch_constituents_sync, index_name)
|
|
logger.info(f"Retrieved {result['count']} constituents for {index_name}")
|
|
return result
|
|
|
|
|
|
index_constituents_service = IndexConstituentsService()
|