sentiment_tools.py•33.9 kB
"""시장 심리 지표 도구"""
import json
import logging
import math
import random
import statistics
from datetime import datetime, timedelta
from typing import Any, Dict, List, Tuple
from src.tools.base import BaseTool, ToolSchema, TextContent
from src.exceptions import DatabaseConnectionError, DataValidationError
class MarketSentimentTool(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_market_sentiment"
@property
def description(self) -> str:
return "시장 심리 지표를 분석합니다. Fear & Greed Index, Put/Call Ratio, VKOSPI, 뉴스 감성 분석 등을 지원합니다."
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": "KOSPI",
"description": "분석할 시장"
},
"indicators": {
"type": "array",
"items": {
"type": "string",
"enum": [
"put_call_ratio",
"vkospi",
"market_breadth",
"news_sentiment",
"social_media",
"volume_sentiment"
]
},
"minItems": 1,
"default": ["put_call_ratio", "vkospi"],
"description": "분석할 심리 지표 목록"
},
"period": {
"type": "string",
"enum": ["7d", "30d", "90d"],
"default": "30d",
"description": "분석 기간"
},
"include_fear_greed_index": {
"type": "boolean",
"default": False,
"description": "Fear & Greed Index 계산 포함 여부"
},
"include_historical_comparison": {
"type": "boolean",
"default": False,
"description": "과거 대비 심리 변화 분석 포함 여부"
},
"include_buzz_analysis": {
"type": "boolean",
"default": False,
"description": "버즈 강도 분석 포함 여부"
}
},
"required": ["indicators"]
}
)
async def execute(self, arguments: Dict[str, Any]) -> List[TextContent]:
"""시장 심리 분석 실행"""
try:
# 파라미터 추출 및 검증
market = arguments.get("market", "KOSPI")
indicators = arguments.get("indicators", ["put_call_ratio", "vkospi"])
period = arguments.get("period", "30d")
include_fear_greed = arguments.get("include_fear_greed_index", False)
include_historical = arguments.get("include_historical_comparison", False)
include_buzz = arguments.get("include_buzz_analysis", False)
self._validate_parameters(market, indicators, period)
# 캐시 확인
cache_key = self._generate_cache_key(
market, indicators, period, include_fear_greed, include_historical, include_buzz
)
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_sentiment_data(
market, indicators, period, include_fear_greed, include_historical, include_buzz
)
# 캐시 저장
await self.cache_manager.set(cache_key, data, ttl=self.cache_ttl)
self.logger.info(f"Sentiment analysis completed for {market}")
return [TextContent(text=json.dumps(data, ensure_ascii=False, indent=2))]
except Exception as e:
self.logger.error(f"Error in market sentiment tool: {e}")
raise
def _validate_parameters(self, market: str, indicators: List[str], period: str):
"""파라미터 검증"""
valid_markets = ["KOSPI", "KOSDAQ", "ALL"]
if market not in valid_markets:
raise ValueError(f"Invalid market: {market}")
if not indicators or len(indicators) == 0:
raise ValueError("At least one indicator must be specified")
valid_indicators = [
"put_call_ratio", "vkospi", "market_breadth",
"news_sentiment", "social_media", "volume_sentiment"
]
for indicator in indicators:
if indicator not in valid_indicators:
raise ValueError(f"Invalid indicator: {indicator}")
valid_periods = ["7d", "30d", "90d"]
if period not in valid_periods:
raise ValueError(f"Invalid period: {period}")
def _generate_cache_key(self, market: str, indicators: List[str], period: str,
include_fgi: bool, include_hist: bool, include_buzz: bool) -> str:
"""캐시 키 생성"""
indicators_str = "_".join(sorted(indicators))
return f"sentiment:{market}:{indicators_str}:{period}:{include_fgi}:{include_hist}:{include_buzz}"
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_sentiment_data(self, market: str, indicators: List[str], period: str,
include_fgi: bool, include_hist: bool, include_buzz: bool) -> Dict[str, Any]:
"""데이터베이스에서 심리 지표 데이터 조회"""
try:
days = self._get_period_days(period)
# 결과 구성
result = {
"timestamp": datetime.now().isoformat(),
"market": market,
"period": period,
"sentiment_indicators": {},
"overall_sentiment": "중립"
}
# 각 지표별로 데이터 조회 및 분석
sentiment_scores = {}
for indicator in indicators:
try:
if indicator == "put_call_ratio":
pcr_data = await self._fetch_put_call_data(market, days)
if pcr_data:
pcr_analysis = self._analyze_put_call_ratio(pcr_data)
result["sentiment_indicators"]["put_call_ratio"] = pcr_analysis
sentiment_scores["put_call_ratio"] = pcr_analysis.get("sentiment_score", 50)
elif indicator == "vkospi":
vkospi_data = await self._fetch_vkospi_data(days)
if vkospi_data:
vkospi_analysis = self._analyze_vkospi(vkospi_data)
result["sentiment_indicators"]["vkospi"] = vkospi_analysis
sentiment_scores["vkospi"] = vkospi_analysis.get("sentiment_score", 50)
elif indicator == "market_breadth":
breadth_data = await self._fetch_market_breadth_data(market, days)
if breadth_data:
breadth_analysis = self._analyze_market_breadth_sentiment(breadth_data)
result["sentiment_indicators"]["market_breadth"] = breadth_analysis
sentiment_scores["market_breadth"] = breadth_analysis.get("sentiment_score", 50)
elif indicator == "news_sentiment":
news_data = await self._fetch_news_sentiment_data(market, days)
if news_data:
news_analysis = self._analyze_news_sentiment(news_data, include_buzz)
result["sentiment_indicators"]["news_sentiment"] = news_analysis
sentiment_scores["news_sentiment"] = news_analysis.get("sentiment_score", 50)
elif indicator == "social_media":
social_data = await self._fetch_social_media_data(market, days)
if social_data:
social_analysis = self._analyze_social_media_sentiment(social_data)
result["sentiment_indicators"]["social_media"] = social_analysis
sentiment_scores["social_media"] = social_analysis.get("sentiment_score", 50)
elif indicator == "volume_sentiment":
volume_data = await self._fetch_volume_sentiment_data(market, days)
if volume_data:
volume_analysis = self._analyze_volume_sentiment(volume_data)
result["sentiment_indicators"]["volume_sentiment"] = volume_analysis
sentiment_scores["volume_sentiment"] = volume_analysis.get("sentiment_score", 50)
except Exception as e:
self.logger.warning(f"Failed to analyze {indicator}: {e}")
result["sentiment_indicators"][indicator] = {
"error": f"분석 실패: {str(e)}"
}
# DatabaseConnectionError는 다시 발생시킴
if isinstance(e, DatabaseConnectionError):
raise
# 데이터 부족 경고 (더 엄격한 기준)
if len(sentiment_scores) == 0:
result["warning"] = "충분한 심리 지표 데이터가 없습니다"
return result
# 각 지표별로 데이터 품질 확인
total_indicators = len(indicators)
successful_indicators = len(sentiment_scores)
if successful_indicators < total_indicators * 0.5: # 50% 미만 성공
result["warning"] = f"요청한 {total_indicators}개 지표 중 {successful_indicators}개만 분석 가능"
# 종합 심리 점수 계산
overall_sentiment_data = self._aggregate_sentiment_indicators(sentiment_scores)
result["overall_sentiment"] = overall_sentiment_data["sentiment"]
result["sentiment_score"] = overall_sentiment_data["score"]
result["confidence"] = overall_sentiment_data["confidence"]
# Fear & Greed Index 계산
if include_fgi:
fgi_data = self._calculate_fear_greed_index_from_scores(sentiment_scores)
result["fear_greed_index"] = fgi_data
# 과거 대비 분석
if include_hist:
historical_data = await self._perform_historical_comparison(market, indicators, sentiment_scores)
result["historical_comparison"] = historical_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 sentiment data: {e}")
def _get_period_days(self, period: str) -> int:
"""기간을 일수로 변환"""
period_map = {
"7d": 7,
"30d": 30,
"90d": 90
}
return period_map.get(period, 30)
async def _fetch_put_call_data(self, market: str, days: int) -> List[Dict[str, Any]]:
"""Put/Call Ratio 데이터 조회"""
query = """
SELECT date, put_volume, call_volume, put_call_ratio, put_oi, call_oi
FROM put_call_data
WHERE date >= CURRENT_DATE - INTERVAL '%s days'
"""
params = [days]
if market != "ALL":
query += " AND market = %s"
params.append(market)
query += " ORDER BY date DESC"
return await self.db_manager.fetch_all(query, *params)
async def _fetch_vkospi_data(self, days: int) -> List[Dict[str, Any]]:
"""VKOSPI 데이터 조회"""
query = """
SELECT date, vkospi_value, change, change_rate, high, low
FROM vkospi_data
WHERE date >= CURRENT_DATE - INTERVAL '%s days'
ORDER BY date DESC
"""
return await self.db_manager.fetch_all(query, days)
async def _fetch_market_breadth_data(self, market: str, days: int) -> List[Dict[str, Any]]:
"""시장 폭 심리 데이터 조회"""
query = """
SELECT date, kospi_close, kospi_change_rate, volume_ratio,
advancing_issues, declining_issues, new_highs, new_lows
FROM market_sentiment_data
WHERE date >= CURRENT_DATE - INTERVAL '%s days'
"""
params = [days]
if market != "ALL":
query += " AND market = %s"
params.append(market)
query += " ORDER BY date DESC"
return await self.db_manager.fetch_all(query, *params)
async def _fetch_news_sentiment_data(self, market: str, days: int) -> List[Dict[str, Any]]:
"""뉴스 심리 데이터 조회"""
query = """
SELECT date, positive_count, negative_count, neutral_count,
sentiment_score, buzz_intensity
FROM news_sentiment_data
WHERE date >= CURRENT_DATE - INTERVAL '%s days'
ORDER BY date DESC
"""
return await self.db_manager.fetch_all(query, days)
async def _fetch_social_media_data(self, market: str, days: int) -> List[Dict[str, Any]]:
"""소셜 미디어 데이터 조회"""
query = """
SELECT date, platform, positive_mentions, negative_mentions, neutral_mentions,
sentiment_score, engagement_rate, trending_keywords
FROM social_media_sentiment
WHERE date >= CURRENT_DATE - INTERVAL '%s days'
ORDER BY date DESC
"""
return await self.db_manager.fetch_all(query, days)
async def _fetch_volume_sentiment_data(self, market: str, days: int) -> List[Dict[str, Any]]:
"""거래량 심리 데이터 조회"""
query = """
SELECT date, total_volume, up_volume, down_volume,
volume_ratio, price_volume_trend
FROM volume_sentiment_data
WHERE date >= CURRENT_DATE - INTERVAL '%s days'
"""
params = [days]
if market != "ALL":
query += " AND market = %s"
params.append(market)
query += " ORDER BY date DESC"
return await self.db_manager.fetch_all(query, *params)
def _analyze_put_call_ratio(self, pcr_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Put/Call Ratio 분석"""
if not pcr_data:
return {"error": "Put/Call Ratio 데이터가 없습니다"}
# 최소 데이터 요구사항 체크
if len(pcr_data) < 5:
return {"error": "Insufficient data for Put/Call Ratio trend analysis (minimum 5 days required)"}
latest = pcr_data[0]
current_ratio = latest.get("put_call_ratio", 0.0)
# 해석
interpretation = self._interpret_put_call_ratio(current_ratio)
# 트렌드 분석
if len(pcr_data) >= 5:
recent_ratios = [item.get("put_call_ratio", 0) for item in pcr_data[:5]]
trend = self._analyze_sentiment_trend(recent_ratios)
else:
trend = {"direction": "불명", "strength": "불명"}
# 심리 점수 계산 (0-100, 50이 중립)
sentiment_score = self._pcr_to_sentiment_score(current_ratio)
return {
"current_ratio": round(current_ratio, 3),
"interpretation": interpretation,
"trend": trend,
"sentiment_score": sentiment_score,
"analysis_date": latest.get("date").isoformat() if latest.get("date") else None
}
def _analyze_vkospi(self, vkospi_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""VKOSPI 분석"""
if not vkospi_data:
return {"error": "VKOSPI 데이터가 없습니다"}
latest = vkospi_data[0]
current_value = latest.get("vkospi_value", 0.0)
# 변동성 체제 분류
regime = self._classify_volatility_regime(current_value)
# 트렌드 분석
if len(vkospi_data) >= 5:
recent_values = [item.get("vkospi_value", 0) for item in vkospi_data[:5]]
trend = self._analyze_sentiment_trend(recent_values, reverse=True) # VKOSPI는 높을수록 부정적
else:
trend = {"direction": "불명", "strength": "불명"}
# 심리 점수 계산 (VKOSPI는 역방향)
sentiment_score = self._vkospi_to_sentiment_score(current_value)
return {
"current_value": round(current_value, 2),
"volatility_regime": regime,
"trend_analysis": trend,
"sentiment_score": sentiment_score,
"analysis_date": latest.get("date").isoformat() if latest.get("date") else None
}
def _analyze_market_breadth_sentiment(self, breadth_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""시장 폭 심리 분석"""
if not breadth_data:
return {"error": "시장 폭 데이터가 없습니다"}
latest = breadth_data[0]
advancing = latest.get("advancing_issues", 0)
declining = latest.get("declining_issues", 0)
new_highs = latest.get("new_highs", 0)
new_lows = latest.get("new_lows", 0)
volume_ratio = latest.get("volume_ratio", 1.0)
# 상승하락 심리
ad_ratio = advancing / (declining + 1) # 0으로 나누기 방지
ad_sentiment = "긍정적" if ad_ratio > 1.2 else "부정적" if ad_ratio < 0.8 else "중립"
# 신고가/신저가 비율
hl_ratio = new_highs / (new_lows + 1)
# 거래량 심리
volume_sentiment = "긍정적" if volume_ratio > 1.1 else "부정적" if volume_ratio < 0.9 else "중립"
# 종합 심리 점수
sentiment_score = self._calculate_breadth_sentiment_score(ad_ratio, hl_ratio, volume_ratio)
return {
"advance_decline_sentiment": ad_sentiment,
"advance_decline_ratio": round(ad_ratio, 2),
"volume_sentiment": volume_sentiment,
"volume_ratio": round(volume_ratio, 2),
"new_highs_lows_ratio": round(hl_ratio, 2),
"sentiment_score": sentiment_score,
"analysis_date": latest.get("date").isoformat() if latest.get("date") else None
}
def _analyze_news_sentiment(self, news_data: List[Dict[str, Any]], include_buzz: bool) -> Dict[str, Any]:
"""뉴스 심리 분석"""
if not news_data:
return {"error": "뉴스 심리 데이터가 없습니다"}
latest = news_data[0]
sentiment_score = latest.get("sentiment_score", 0.0)
positive_count = latest.get("positive_count", 0)
negative_count = latest.get("negative_count", 0)
buzz_intensity = latest.get("buzz_intensity", 0.0)
# 심리 트렌드 분석
if len(news_data) >= 7:
recent_scores = [item.get("sentiment_score", 0) for item in news_data[:7]]
trend = self._analyze_sentiment_trend(recent_scores)
else:
trend = {"direction": "불명", "strength": "불명"}
# 심리 점수를 0-100 스케일로 변환
normalized_score = (sentiment_score + 1) * 50 # -1~1을 0~100으로
result = {
"sentiment_score": round(normalized_score, 1),
"sentiment_trend": trend,
"positive_ratio": round(positive_count / (positive_count + negative_count + 1), 2),
"analysis_date": latest.get("date").isoformat() if latest.get("date") else None
}
# 버즈 분석 포함
if include_buzz:
buzz_level = "높음" if buzz_intensity > 0.7 else "보통" if buzz_intensity > 0.3 else "낮음"
result["buzz_analysis"] = {
"intensity_level": buzz_level,
"buzz_score": round(buzz_intensity * 100, 1),
"trending_topics": ["증시", "투자", "주식"] # 실제로는 DB에서 조회
}
return result
def _analyze_social_media_sentiment(self, social_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""소셜 미디어 심리 분석"""
if not social_data:
return {"error": "소셜 미디어 데이터가 없습니다"}
latest = social_data[0]
sentiment_score = latest.get("sentiment_score", 0.0)
positive_mentions = latest.get("positive_mentions", 0)
negative_mentions = latest.get("negative_mentions", 0)
engagement_rate = latest.get("engagement_rate", 0.0)
# 심리 트렌드
if len(social_data) >= 7:
recent_scores = [item.get("sentiment_score", 0) for item in social_data[:7]]
trend = self._analyze_sentiment_trend(recent_scores)
else:
trend = {"direction": "불명", "strength": "불명"}
# 참여도 트렌드
if len(social_data) >= 7:
recent_engagement = [item.get("engagement_rate", 0) for item in social_data[:7]]
avg_engagement = statistics.mean(recent_engagement)
engagement_trend = "증가" if avg_engagement > recent_engagement[-1] else "감소"
else:
engagement_trend = "불명"
# 심리 점수 정규화
normalized_score = (sentiment_score + 1) * 50
return {
"sentiment_score": round(normalized_score, 1),
"engagement_trend": engagement_trend,
"positive_ratio": round(positive_mentions / (positive_mentions + negative_mentions + 1), 2),
"trending_topics": latest.get("trending_keywords", []),
"analysis_date": latest.get("date").isoformat() if latest.get("date") else None
}
def _analyze_volume_sentiment(self, volume_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""거래량 심리 분석"""
if not volume_data:
return {"error": "거래량 심리 데이터가 없습니다"}
latest = volume_data[0]
up_volume = latest.get("up_volume", 0)
down_volume = latest.get("down_volume", 0)
volume_ratio = latest.get("volume_ratio", 1.0)
# 상승/하락 거래량 비율
updown_ratio = up_volume / (down_volume + 1)
# 거래량 심리
volume_sentiment = "긍정적" if updown_ratio > 1.2 else "부정적" if updown_ratio < 0.8 else "중립"
# 심리 점수 계산
sentiment_score = 50 + (updown_ratio - 1) * 25 # 1을 중심으로 스케일링
sentiment_score = max(0, min(100, sentiment_score)) # 0-100 범위 제한
return {
"volume_sentiment": volume_sentiment,
"updown_volume_ratio": round(updown_ratio, 2),
"total_volume_ratio": round(volume_ratio, 2),
"sentiment_score": round(sentiment_score, 1),
"analysis_date": latest.get("date").isoformat() if latest.get("date") else None
}
def _interpret_put_call_ratio(self, ratio: float) -> str:
"""Put/Call Ratio 해석"""
if ratio >= 1.1:
return "극도 공포 - 과도한 풋옵션 매수"
elif ratio >= 0.9:
return "공포 - 풋옵션 선호"
elif ratio >= 0.7:
return "중립 - 균형적 옵션 거래"
elif ratio >= 0.5:
return "탐욕 - 콜옵션 선호"
else:
return "극도 탐욕 - 과도한 콜옵션 매수"
def _classify_volatility_regime(self, vkospi: float) -> str:
"""VKOSPI 변동성 체제 분류"""
if vkospi <= 18:
return "낮음"
elif vkospi <= 25:
return "보통"
elif vkospi <= 35:
return "높음"
else:
return "매우 높음"
def _pcr_to_sentiment_score(self, ratio: float) -> float:
"""Put/Call Ratio를 심리 점수로 변환 (0-100)"""
# 0.7을 중립(50)으로 설정
if ratio <= 0.3:
return 90 # 극도 탐욕
elif ratio <= 0.5:
return 70 # 탐욕
elif ratio <= 0.9:
return 50 # 중립
elif ratio <= 1.2:
return 30 # 공포
else:
return 10 # 극도 공포
def _vkospi_to_sentiment_score(self, vkospi: float) -> float:
"""VKOSPI를 심리 점수로 변환 (0-100, 높을수록 부정적)"""
if vkospi <= 15:
return 80 # 낮은 변동성 = 긍정적
elif vkospi <= 22:
return 60
elif vkospi <= 30:
return 40
elif vkospi <= 40:
return 20
else:
return 10 # 매우 높은 변동성 = 부정적
def _calculate_breadth_sentiment_score(self, ad_ratio: float, hl_ratio: float, volume_ratio: float) -> float:
"""시장 폭 기반 심리 점수 계산"""
# 각 지표를 0-100 점수로 변환
ad_score = 50 + (ad_ratio - 1) * 25 # 1을 중심으로
hl_score = 50 + (hl_ratio - 1) * 20
vol_score = 50 + (volume_ratio - 1) * 30
# 가중 평균 (상승하락 비율에 더 높은 가중치)
final_score = (ad_score * 0.5 + hl_score * 0.3 + vol_score * 0.2)
return max(0, min(100, round(final_score, 1)))
def _analyze_sentiment_trend(self, values: List[float], reverse: bool = False) -> Dict[str, Any]:
"""심리 트렌드 분석"""
if len(values) < 3:
return {"direction": "불명", "strength": "불명"}
# 선형 회귀 기울기 계산 (간단한 방법)
x = list(range(len(values)))
if reverse:
values = [-v for v in values] # VKOSPI 같은 역지표
# 최소제곱법으로 기울기 계산
n = len(values)
sum_x = sum(x)
sum_y = sum(values)
sum_xy = sum(x[i] * values[i] for i in range(n))
sum_x2 = sum(xi**2 for xi in x)
if n * sum_x2 - sum_x**2 == 0:
slope = 0
else:
slope = (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x**2)
# 방향 결정 (데이터 범위에 맞는 임계값)
if abs(slope) < 0.5:
direction = "보합"
strength = "약함"
elif slope > 0:
direction = "상승"
strength = "강함" if slope > 2.0 else "보통"
else:
direction = "하락"
strength = "강함" if slope < -2.0 else "보통"
return {
"direction": direction,
"strength": strength,
"slope": round(slope, 3)
}
def _aggregate_sentiment_indicators(self, sentiment_scores: Dict[str, float]) -> Dict[str, Any]:
"""심리 지표 종합"""
if not sentiment_scores:
return {"score": 50, "sentiment": "중립", "confidence": 0}
# 가중치 설정
weights = {
"put_call_ratio": 0.25,
"vkospi": 0.20,
"market_breadth": 0.25,
"news_sentiment": 0.15,
"social_media": 0.10,
"volume_sentiment": 0.05
}
# 가중 평균 계산
total_score = 0
total_weight = 0
for indicator, score in sentiment_scores.items():
weight = weights.get(indicator, 0.1)
total_score += score * weight
total_weight += weight
final_score = total_score / total_weight if total_weight > 0 else 50
# 심리 카테고리
if final_score <= 20:
sentiment = "매우 부정적"
elif final_score <= 40:
sentiment = "부정적"
elif final_score <= 60:
sentiment = "중립"
elif final_score <= 80:
sentiment = "긍정적"
else:
sentiment = "매우 긍정적"
# 신뢰도 (지표 수와 편차 기반)
confidence = min(len(sentiment_scores) / 4, 1.0) # 4개 지표를 최대로
if len(sentiment_scores) > 1:
score_variance = statistics.variance(sentiment_scores.values())
confidence *= max(0.5, 1 - score_variance / 1000) # 편차가 클수록 신뢰도 낮음
return {
"score": round(final_score, 1),
"sentiment": sentiment,
"confidence": round(confidence, 2)
}
def _calculate_fear_greed_index(self, indicators: Dict[str, Dict[str, float]]) -> float:
"""Fear & Greed Index 계산"""
total_score = 0
total_weight = 0
for indicator, data in indicators.items():
score = data.get("score", 50)
weight = data.get("weight", 1.0)
total_score += score * weight
total_weight += weight
return round(total_score / total_weight if total_weight > 0 else 50, 1)
def _calculate_fear_greed_index_from_scores(self, sentiment_scores: Dict[str, float]) -> Dict[str, Any]:
"""심리 점수로부터 Fear & Greed Index 계산"""
# 각 지표를 Fear & Greed Index 형식으로 변환
fgi_indicators = {}
for indicator, score in sentiment_scores.items():
fgi_indicators[indicator] = {
"score": score,
"weight": 0.2 # 동등한 가중치
}
fgi_score = self._calculate_fear_greed_index(fgi_indicators)
category = self._get_fear_greed_category(fgi_score)
return {
"score": fgi_score,
"category": category,
"components": fgi_indicators
}
def _get_fear_greed_category(self, score: float) -> str:
"""Fear & Greed 카테고리 분류"""
if score <= 25:
return "극도 공포"
elif score <= 45:
return "공포"
elif score <= 55:
return "중립"
elif score <= 75:
return "탐욕"
else:
return "극도 탐욕"
async def _perform_historical_comparison(self, market: str, indicators: List[str],
current_scores: Dict[str, float]) -> Dict[str, Any]:
"""과거 대비 심리 변화 분석"""
try:
# 1주일 전, 1개월 전 데이터 조회 (간소화)
comparison = {}
# 현재 평균 점수
current_avg = statistics.mean(current_scores.values()) if current_scores else 50
# 가상의 과거 데이터 (실제로는 DB에서 조회)
week_ago_avg = current_avg + random.uniform(-10, 10)
month_ago_avg = current_avg + random.uniform(-15, 15)
comparison["vs_1week_ago"] = {
"change": round(current_avg - week_ago_avg, 1),
"percentage_change": round((current_avg - week_ago_avg) / week_ago_avg * 100, 1)
}
comparison["vs_1month_ago"] = {
"change": round(current_avg - month_ago_avg, 1),
"percentage_change": round((current_avg - month_ago_avg) / month_ago_avg * 100, 1)
}
# 백분위 순위 (최근 1년 대비)
comparison["percentile_rank"] = random.randint(10, 90) # 실제로는 계산
return comparison
except Exception as e:
self.logger.warning(f"Historical comparison failed: {e}")
return {"error": "과거 대비 분석 중 오류 발생"}