"""섹터 분석 도구"""
import json
import logging
import statistics
from datetime import datetime, timedelta
from typing import Any, Dict, List
from src.tools.base import BaseTool, ToolSchema, TextContent
from src.exceptions import DatabaseConnectionError, DataValidationError
class SectorAnalysisTool(BaseTool):
"""섹터별 분석 도구"""
def __init__(self, db_manager, cache_manager):
super().__init__(db_manager, cache_manager)
self.logger = logging.getLogger(__name__)
self.cache_ttl = 300 # 5분
@property
def name(self) -> str:
return "get_sector_analysis"
@property
def description(self) -> str:
return "업종별/섹터별 주식 시장 분석을 제공합니다. 섹터 성과, 시가총액, 주요 종목 등을 분석합니다."
def get_tool_definition(self) -> ToolSchema:
"""도구 정의 반환"""
return ToolSchema(
name=self.name,
description=self.description,
inputSchema={
"type": "object",
"properties": {
"market": {
"type": "string",
"enum": ["KOSPI", "KOSDAQ", "ALL"],
"default": "ALL",
"description": "조회할 시장"
},
"sector": {
"type": "string",
"default": "ALL",
"description": "조회할 섹터 (예: IT, 금융, 바이오, ALL)"
},
"sort_by": {
"type": "string",
"enum": ["market_cap", "change_rate", "volume", "strength"],
"default": "market_cap",
"description": "정렬 기준"
},
"include_stocks": {
"type": "boolean",
"default": False,
"description": "주요 종목 정보 포함 여부"
},
"include_relative_strength": {
"type": "boolean",
"default": False,
"description": "상대 강도 분석 포함 여부"
},
"include_rotation_analysis": {
"type": "boolean",
"default": False,
"description": "섹터 로테이션 분석 포함 여부"
}
},
"required": []
}
)
async def execute(self, arguments: Dict[str, Any]) -> List[TextContent]:
"""섹터 분석 데이터 조회 및 분석"""
try:
# 파라미터 추출 및 검증
market = arguments.get("market", "ALL")
sector = arguments.get("sector", "ALL")
sort_by = arguments.get("sort_by", "market_cap")
include_stocks = arguments.get("include_stocks", False)
include_relative_strength = arguments.get("include_relative_strength", False)
include_rotation_analysis = arguments.get("include_rotation_analysis", False)
self._validate_parameters(market, sort_by)
# 캐시 확인
cache_key = self._generate_cache_key(market, sector, sort_by, include_stocks, include_relative_strength, include_rotation_analysis)
cached_data = await self.cache_manager.get(cache_key)
if cached_data and self._is_data_fresh(cached_data):
self.logger.info(f"Cache hit for {cache_key}")
return [TextContent(text=json.dumps(cached_data, ensure_ascii=False))]
# 데이터베이스에서 데이터 조회
data = await self._fetch_sector_data(market, sector, sort_by, include_stocks, include_relative_strength, include_rotation_analysis)
# 캐시 저장
await self.cache_manager.set(cache_key, data, ttl=self.cache_ttl)
self.logger.info(f"Sector analysis data fetched for {market}/{sector}")
return [TextContent(text=json.dumps(data, ensure_ascii=False, indent=2))]
except Exception as e:
self.logger.error(f"Error in sector analysis tool: {e}")
raise
def _validate_parameters(self, market: str, sort_by: str):
"""파라미터 검증"""
valid_markets = ["KOSPI", "KOSDAQ", "ALL"]
valid_sort_by = ["market_cap", "change_rate", "volume", "strength"]
if market not in valid_markets:
raise ValueError(f"Invalid market: {market}")
if sort_by not in valid_sort_by:
raise ValueError(f"Invalid sort_by: {sort_by}")
def _generate_cache_key(self, market: str, sector: str, sort_by: str, include_stocks: bool, include_rs: bool, include_rotation: bool) -> str:
"""캐시 키 생성"""
return f"sector_analysis:{market}:{sector}:{sort_by}:{include_stocks}:{include_rs}:{include_rotation}"
def _is_data_fresh(self, data: Dict[str, Any]) -> bool:
"""데이터 신선도 확인"""
if "timestamp" not in data:
return False
try:
timestamp = datetime.fromisoformat(data["timestamp"])
return datetime.now() - timestamp < timedelta(minutes=5)
except (ValueError, TypeError):
return False
async def _fetch_sector_data(self, market: str, sector: str, sort_by: str, include_stocks: bool, include_rs: bool, include_rotation: bool) -> Dict[str, Any]:
"""데이터베이스에서 섹터 분석 데이터 조회"""
try:
# 쿼리 구성
query = """
SELECT sector_name, sector_code, market_cap, total_volume, change_rate,
advancing_stocks, declining_stocks, unchanged_stocks, total_stocks,
top_performer, top_performer_change, worst_performer, worst_performer_change,
date, timestamp
FROM sector_performance
WHERE DATE(date) = CURRENT_DATE
"""
params = []
# 시장 필터
if market != "ALL":
query += " AND market = %s"
params.append(market)
# 섹터 필터
if sector != "ALL":
query += " AND (sector_name = %s OR sector_code = %s)"
params.extend([sector, sector])
# 정렬 추가
sort_column = self._get_sort_column(sort_by)
query += f" ORDER BY {sort_column} DESC"
# 데이터 조회
sector_data = await self.db_manager.fetch_all(query, *params)
# 정렬 적용
sorted_data = self._sort_sectors(sector_data, sort_by)
# 결과 구성
result = {
"timestamp": datetime.now().isoformat(),
"market": market,
"sector": sector,
"sort_by": sort_by,
"sector_analysis": [self._format_sector_data(data) for data in sorted_data],
"market_summary": self._calculate_market_summary(sorted_data)
}
# 빈 데이터 처리
if not sector_data:
result["message"] = "요청한 섹터의 데이터가 없습니다"
result["sector_analysis"] = []
# 주요 종목 정보 포함
if include_stocks and sector_data:
await self._add_major_stocks_info(result, sorted_data)
# 상대 강도 분석 포함
if include_rs and sector_data:
result["relative_strength"] = self._analyze_relative_strength(sorted_data)
# 섹터 로테이션 분석 포함
if include_rotation and sector_data:
rotation_data = await self._analyze_sector_rotation(sorted_data)
result["rotation_analysis"] = rotation_data
return result
except Exception as e:
self.logger.error(f"Database query failed: {e}")
if isinstance(e, DatabaseConnectionError):
raise
raise DatabaseConnectionError(f"Failed to fetch sector data: {e}")
def _get_sort_column(self, sort_by: str) -> str:
"""정렬 컬럼 매핑"""
column_map = {
"market_cap": "market_cap",
"change_rate": "change_rate",
"volume": "total_volume",
"strength": "change_rate" # 임시로 change_rate 사용
}
return column_map.get(sort_by, "market_cap")
def _sort_sectors(self, data: List[Dict[str, Any]], sort_by: str) -> List[Dict[str, Any]]:
"""섹터 데이터 정렬"""
if not data:
return data
if sort_by == "market_cap":
return sorted(data, key=lambda x: x.get("market_cap", 0), reverse=True)
elif sort_by == "change_rate":
return sorted(data, key=lambda x: x.get("change_rate", 0), reverse=True)
elif sort_by == "volume":
return sorted(data, key=lambda x: x.get("total_volume", 0), reverse=True)
elif sort_by == "strength":
# 섹터 강도로 정렬 (상승종목비율 + 변화율 고려)
return sorted(data, key=lambda x: self._calculate_sector_strength_score(x), reverse=True)
else:
return data
def _calculate_sector_strength_score(self, sector_data: Dict[str, Any]) -> float:
"""섹터 강도 점수 계산"""
advancing = sector_data.get("advancing_stocks", 0)
total = sector_data.get("total_stocks", 1)
change_rate = sector_data.get("change_rate", 0)
advance_ratio = advancing / total if total > 0 else 0
return advance_ratio * 0.6 + (change_rate / 10.0) * 0.4
def _format_sector_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""섹터 데이터 포맷팅"""
advancing = data.get("advancing_stocks", 0)
declining = data.get("declining_stocks", 0)
unchanged = data.get("unchanged_stocks", 0)
change_rate = data.get("change_rate", 0)
# 섹터 강도 계산
strength = self._calculate_sector_strength(advancing, declining, unchanged, change_rate)
return {
"sector_name": data.get("sector_name"),
"sector_code": data.get("sector_code"),
"market_cap": int(data.get("market_cap", 0)),
"market_cap_formatted": self._format_market_cap(data.get("market_cap", 0)),
"total_volume": int(data.get("total_volume", 0)),
"change_rate": float(data.get("change_rate", 0)),
"advancing_stocks": int(advancing),
"declining_stocks": int(declining),
"unchanged_stocks": int(unchanged),
"total_stocks": int(data.get("total_stocks", 0)),
"advance_decline_ratio": round(advancing / declining if declining > 0 else float('inf'), 2),
"sector_strength": strength,
"top_performer": data.get("top_performer"),
"top_performer_change": float(data.get("top_performer_change", 0)),
"worst_performer": data.get("worst_performer"),
"worst_performer_change": float(data.get("worst_performer_change", 0)),
"date": data.get("date").isoformat() if data.get("date") else None,
"timestamp": data.get("timestamp").isoformat() if data.get("timestamp") else None
}
def _calculate_sector_strength(self, advancing: int, declining: int, unchanged: int, change_rate: float) -> str:
"""섹터 강도 계산"""
total = advancing + declining + unchanged
if total == 0:
return "알 수 없음"
advance_ratio = advancing / total
# 상승 종목 비율과 섹터 변화율을 종합 고려
if advance_ratio >= 0.59 and change_rate >= 4.0:
return "매우 강함"
elif advance_ratio >= 0.5 and change_rate >= 1.0:
return "강함"
elif advance_ratio >= 0.3 and change_rate >= -1.0:
return "보통"
elif advance_ratio >= 0.2 and change_rate >= -3.0:
return "약함"
else:
return "매우 약함"
def _format_market_cap(self, market_cap: float) -> str:
"""시가총액 포맷팅"""
if market_cap >= 1000000000000: # 1조 이상
trillion = market_cap / 1000000000000
return f"{trillion:.1f}조"
elif market_cap >= 100000000000: # 1000억 이상
hundred_billion = market_cap / 100000000000
return f"{hundred_billion:.0f}00억"
elif market_cap >= 100000000: # 1억 이상
hundred_million = market_cap / 100000000
return f"{hundred_million:.0f}억"
else:
return f"{int(market_cap):,}원"
def _calculate_market_summary(self, sector_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""시장 요약 계산"""
if not sector_data:
return {}
total_market_cap = sum(data.get("market_cap", 0) for data in sector_data)
positive_sectors = len([s for s in sector_data if s.get("change_rate", 0) > 0])
negative_sectors = len([s for s in sector_data if s.get("change_rate", 0) < 0])
avg_change_rate = statistics.mean([s.get("change_rate", 0) for s in sector_data])
return {
"sector_count": len(sector_data),
"total_market_cap": int(total_market_cap),
"total_market_cap_formatted": self._format_market_cap(total_market_cap),
"positive_sectors": positive_sectors,
"negative_sectors": negative_sectors,
"neutral_sectors": len(sector_data) - positive_sectors - negative_sectors,
"avg_change_rate": round(avg_change_rate, 2),
"market_sentiment": "상승" if avg_change_rate > 0.5 else "하락" if avg_change_rate < -0.5 else "보합"
}
async def _add_major_stocks_info(self, result: Dict[str, Any], sector_data: List[Dict[str, Any]]):
"""주요 종목 정보 추가"""
try:
# 각 섹터별로 주요 종목 조회
major_stocks_queries = []
for sector in sector_data[:5]: # 상위 5개 섹터만
sector_code = sector.get("sector_code")
if sector_code:
query = """
SELECT stock_code, stock_name, current_price, change_rate, volume, market_cap
FROM stock_daily_data
WHERE sector_code = %s AND DATE(date) = CURRENT_DATE
ORDER BY market_cap DESC
LIMIT 3
"""
major_stocks_queries.append((query, sector_code))
# 배치 쿼리 실행
if major_stocks_queries:
queries = [q[0] for q in major_stocks_queries]
params_list = [[q[1]] for q in major_stocks_queries]
stocks_results = await self.db_manager.fetch_many(queries, params_list)
# 결과를 섹터별로 매핑
for i, sector in enumerate(result["sector_analysis"][:len(stocks_results)]):
if i < len(stocks_results):
sector["major_stocks"] = [
{
"stock_code": stock.get("stock_code"),
"stock_name": stock.get("stock_name"),
"current_price": int(stock.get("current_price", 0)),
"change_rate": float(stock.get("change_rate", 0)),
"volume": int(stock.get("volume", 0)),
"market_cap": int(stock.get("market_cap", 0))
}
for stock in stocks_results[i]
]
except Exception as e:
self.logger.warning(f"Failed to fetch major stocks info: {e}")
def _analyze_relative_strength(self, sector_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""상대 강도 분석"""
if len(sector_data) < 2:
return {"error": "상대 강도 분석을 위한 충분한 데이터가 없습니다"}
# 변화율 기준으로 정렬
sorted_by_performance = sorted(sector_data, key=lambda x: x.get("change_rate", 0), reverse=True)
strongest = sorted_by_performance[0]
weakest = sorted_by_performance[-1]
return {
"strongest_sector": {
"name": strongest.get("sector_name"),
"change_rate": strongest.get("change_rate"),
"advance_ratio": strongest.get("advancing_stocks", 0) / max(strongest.get("total_stocks", 1), 1)
},
"weakest_sector": {
"name": weakest.get("sector_name"),
"change_rate": weakest.get("change_rate"),
"advance_ratio": weakest.get("advancing_stocks", 0) / max(weakest.get("total_stocks", 1), 1)
},
"performance_spread": round(strongest.get("change_rate", 0) - weakest.get("change_rate", 0), 2)
}
async def _analyze_sector_rotation(self, current_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""섹터 로테이션 분석"""
try:
# 과거 데이터 조회 (1주일 전)
historical_query = """
SELECT sector_name, change_rate
FROM sector_performance
WHERE DATE(date) = CURRENT_DATE - INTERVAL '7 days'
"""
historical_data = await self.db_manager.fetch_all(historical_query)
if not historical_data:
return {"error": "로테이션 분석을 위한 과거 데이터가 없습니다"}
# 현재와 과거 성과 비교
momentum_leaders = []
momentum_laggards = []
# 섹터별 모멘텀 계산
historical_dict = {h.get("sector_name"): h.get("change_rate", 0) for h in historical_data}
for current in current_data:
sector_name = current.get("sector_name")
current_rate = current.get("change_rate", 0)
historical_rate = historical_dict.get(sector_name, 0)
momentum = current_rate - historical_rate
if momentum > 1.0: # 1% 이상 개선
momentum_leaders.append({
"sector": sector_name,
"momentum": round(momentum, 2),
"current_rate": current_rate,
"historical_rate": historical_rate
})
elif momentum < -1.0: # 1% 이상 악화
momentum_laggards.append({
"sector": sector_name,
"momentum": round(momentum, 2),
"current_rate": current_rate,
"historical_rate": historical_rate
})
return {
"momentum_leaders": sorted(momentum_leaders, key=lambda x: x["momentum"], reverse=True)[:3],
"momentum_laggards": sorted(momentum_laggards, key=lambda x: x["momentum"])[:3],
"analysis_period": "1주일"
}
except Exception as e:
self.logger.warning(f"Sector rotation analysis failed: {e}")
return {"error": f"로테이션 분석 중 오류 발생: {e}"}
