311 lines
12 KiB
Python
311 lines
12 KiB
Python
from anthropic import AsyncAnthropic
|
|
from config import settings
|
|
from utils import _dumps
|
|
from typing import Dict, Any, List
|
|
import json
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# 캐싱 공통 컨텍스트 — 전략 규칙을 한 번만 전송하고 이후 호출에서 캐시로 재사용.
|
|
# Claude Sonnet 기준 캐싱 최소 임계값: 1024 토큰.
|
|
_CACHED_CONTEXT = """[자동 주식 매매 시스템 — 리스크 규칙]
|
|
포지션 사이징: 종목당 총자본 3~4%, 최대 보유 20~25개, 레버리지 ETF 최대 20%.
|
|
손절(SL): 안정종목 -6%, 변동성종목 -10%, ATR 우선(진입가 - 2.5*ATR).
|
|
익절(TP): +2R 도달 시 50% 분할매도, 나머지는 고점-(1.5*ATR) 트레일링스탑.
|
|
MDD -10%: 신규진입 중단, 포지션 30% 축소, 단기전략 중단.
|
|
MDD -20%: 전 포지션 50% 강제청산 (레버리지 우선).
|
|
가격필터: 국내 5,000원 초과, 미국 $4 초과 (ETF·리츠 예외).
|
|
상관관계: 섹터당 최대 3종목, 상관계수 0.7 초과 진입 거부.
|
|
시장국면: SIDEWAYS(ADX<20) 진입 금지.
|
|
전략A(평균회귀): 상승장 RSI<40·BB하단, 하락장 RSI<30·BB하단·반전캔들, 거래량>20MA.
|
|
전략B(코스닥스캐너): 총자본 25%, 점수=0.5*RSI+0.3*BB+0.2*Vol 상위 20개.
|
|
단기전략(15분): RSI<30·BB하단·5MA돌파·거래량>1.3배, MDD -10% 미만 비활성.
|
|
"""
|
|
|
|
|
|
class ClaudeClient:
|
|
"""Claude AI 클라이언트"""
|
|
|
|
def __init__(self):
|
|
self.client = AsyncAnthropic(api_key=settings.ANTHROPIC_API_KEY)
|
|
self.model = settings.CLAUDE_MODEL
|
|
|
|
async def analyze(
|
|
self,
|
|
system_prompt: str,
|
|
user_message: str,
|
|
tools: List[Dict[str, Any]] = None
|
|
) -> Dict[str, Any]:
|
|
"""Claude를 통한 분석 실행.
|
|
|
|
시스템 프롬프트는 두 블록으로 구성:
|
|
1) _CACHED_CONTEXT — cache_control 적용, 반복 호출 시 캐시에서 재사용.
|
|
2) 메서드별 role 지시문 — 캐싱 없이 매번 전송.
|
|
"""
|
|
try:
|
|
messages = [
|
|
{"role": "user", "content": user_message}
|
|
]
|
|
|
|
params = {
|
|
"model": self.model,
|
|
"max_tokens": settings.MAX_TOKENS,
|
|
"temperature": settings.TEMPERATURE,
|
|
"system": [
|
|
{
|
|
"type": "text",
|
|
"text": _CACHED_CONTEXT,
|
|
"cache_control": {"type": "ephemeral"},
|
|
},
|
|
{
|
|
"type": "text",
|
|
"text": system_prompt,
|
|
},
|
|
],
|
|
"messages": messages,
|
|
}
|
|
|
|
if tools:
|
|
params["tools"] = tools
|
|
|
|
response = await self.client.messages.create(**params)
|
|
|
|
# 응답 파싱
|
|
usage = response.usage
|
|
cache_read = getattr(usage, "cache_read_input_tokens", 0) or 0
|
|
cache_write = getattr(usage, "cache_creation_input_tokens", 0) or 0
|
|
if cache_read:
|
|
logger.debug(f"[cache] hit={cache_read} write={cache_write} input={usage.input_tokens} output={usage.output_tokens}")
|
|
result = {
|
|
"content": [],
|
|
"stop_reason": response.stop_reason,
|
|
"usage": {
|
|
"input_tokens": usage.input_tokens,
|
|
"output_tokens": usage.output_tokens,
|
|
"cache_read_tokens": cache_read,
|
|
"cache_write_tokens": cache_write,
|
|
},
|
|
}
|
|
|
|
for block in response.content:
|
|
if block.type == "text":
|
|
result["content"].append({
|
|
"type": "text",
|
|
"text": block.text
|
|
})
|
|
elif block.type == "tool_use":
|
|
result["content"].append({
|
|
"type": "tool_use",
|
|
"id": block.id,
|
|
"name": block.name,
|
|
"input": block.input
|
|
})
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Claude analysis error: {e}")
|
|
raise
|
|
|
|
async def validate_trading_rule(
|
|
self,
|
|
rule: Dict[str, Any],
|
|
market_data: Dict[str, Any],
|
|
news_data: List[Dict[str, Any]] = None
|
|
) -> Dict[str, Any]:
|
|
"""트레이딩 규칙 검증"""
|
|
system_prompt = """주식 트레이딩 규칙 검증 전문가. 규칙/시장/뉴스를 분석하여 JSON으로 응답.
|
|
필수 필드: decision(BUY/SELL/HOLD), confidence(0-1), risk_level, recommendation.
|
|
악재(유상증자,감자,소송,실적악화) 시 confidence 낮추고 경고 포함."""
|
|
|
|
user_message = f"규칙:{_dumps(rule)}\n시장:{_dumps(market_data)}"
|
|
|
|
if news_data:
|
|
user_message += f"\n뉴스:{_dumps(news_data)}"
|
|
|
|
response = await self.analyze(system_prompt, user_message)
|
|
|
|
# 텍스트 응답 파싱
|
|
for content in response["content"]:
|
|
if content["type"] == "text":
|
|
try:
|
|
# JSON 응답 파싱 시도
|
|
return json.loads(content["text"])
|
|
except json.JSONDecodeError:
|
|
# JSON이 아닌 경우 기본 구조로 반환
|
|
return {
|
|
"decision": "HOLD",
|
|
"confidence": 0.5,
|
|
"analysis": content["text"]
|
|
}
|
|
|
|
return {
|
|
"decision": "HOLD",
|
|
"confidence": 0.0,
|
|
"analysis": "분석 실패"
|
|
}
|
|
|
|
async def analyze_market(
|
|
self,
|
|
stock_code: str,
|
|
market_data: Dict[str, Any],
|
|
news: List[Dict[str, Any]] = None
|
|
) -> Dict[str, Any]:
|
|
"""시장 분석"""
|
|
system_prompt = """주식 시장 분석 전문가. JSON으로 응답.
|
|
필수 필드: trend(UP/DOWN/SIDEWAYS), sentiment(POSITIVE/NEGATIVE/NEUTRAL), recommendation(BUY/SELL/HOLD), confidence(0-1), analysis(간략설명)."""
|
|
|
|
news_limit = settings.NEWS_LIMIT if settings.TOKEN_SAVE_MODE else 5
|
|
news_text = ""
|
|
if news:
|
|
news_text = "\n뉴스:" + ",".join([
|
|
item.get('title', '') for item in news[:news_limit]
|
|
])
|
|
|
|
user_message = f"종목:{stock_code}\n시장:{_dumps(market_data)}{news_text}"
|
|
|
|
response = await self.analyze(system_prompt, user_message)
|
|
|
|
for content in response["content"]:
|
|
if content["type"] == "text":
|
|
try:
|
|
return json.loads(content["text"])
|
|
except json.JSONDecodeError:
|
|
return {
|
|
"trend": "SIDEWAYS",
|
|
"sentiment": "NEUTRAL",
|
|
"recommendation": "HOLD",
|
|
"confidence": 0.5,
|
|
"analysis": content["text"]
|
|
}
|
|
|
|
return {
|
|
"trend": "UNKNOWN",
|
|
"sentiment": "NEUTRAL",
|
|
"recommendation": "HOLD",
|
|
"confidence": 0.0
|
|
}
|
|
|
|
async def calculate_fair_price(
|
|
self,
|
|
stock_code: str,
|
|
market_data: Dict[str, Any],
|
|
financial_data: Dict[str, Any] = None
|
|
) -> Dict[str, Any]:
|
|
"""AI 기반 적정 매수/매도 가격 산출"""
|
|
system_prompt = """주식 가격 전략 전문가. JSON으로 응답.
|
|
필수 필드: buy_prices(aggressive,neutral,conservative), sell_prices(target1,target2,target3), stop_loss, strategy, confidence(0-1)."""
|
|
|
|
user_message = f"종목:{stock_code} 현재가:{market_data.get('current_price')} 고가:{market_data.get('high_price')} 저가:{market_data.get('low_price')} 거래량:{market_data.get('volume')} 지표:{_dumps(market_data.get('indicators', {}))}"
|
|
|
|
response = await self.analyze(system_prompt, user_message)
|
|
|
|
for content in response["content"]:
|
|
if content["type"] == "text":
|
|
try:
|
|
return json.loads(content["text"])
|
|
except json.JSONDecodeError:
|
|
return {
|
|
"buy_prices": {"neutral": market_data.get('current_price', 0)},
|
|
"sell_prices": {"target1": market_data.get('current_price', 0) * 1.05},
|
|
"stop_loss": market_data.get('current_price', 0) * 0.97,
|
|
"strategy": content["text"],
|
|
"confidence": 0.5
|
|
}
|
|
|
|
return {
|
|
"error": "분석 실패",
|
|
"confidence": 0.0
|
|
}
|
|
|
|
async def assess_risk(
|
|
self,
|
|
portfolio: Dict[str, Any],
|
|
trade_plan: Dict[str, Any]
|
|
) -> Dict[str, Any]:
|
|
"""리스크 평가"""
|
|
system_prompt = """투자 리스크 관리 전문가. JSON으로 응답.
|
|
필수 필드: risk_level(LOW/MEDIUM/HIGH/CRITICAL), risk_factors, approval(APPROVED/REJECTED), recommendation."""
|
|
|
|
user_message = f"포트폴리오:{_dumps(portfolio)}\n거래계획:{_dumps(trade_plan)}"
|
|
|
|
response = await self.analyze(system_prompt, user_message)
|
|
|
|
for content in response["content"]:
|
|
if content["type"] == "text":
|
|
try:
|
|
return json.loads(content["text"])
|
|
except json.JSONDecodeError:
|
|
return {
|
|
"risk_level": "MEDIUM",
|
|
"approval": "HOLD",
|
|
"confidence": 0.5,
|
|
"analysis": content["text"]
|
|
}
|
|
|
|
return {"error": "Failed to assess risk"}
|
|
|
|
async def optimize_trading_rules(
|
|
self,
|
|
stock_code: str,
|
|
market_data: Dict[str, Any],
|
|
news_data: List[Dict[str, Any]] = None
|
|
) -> Dict[str, Any]:
|
|
"""시장 상황에 맞는 트레이딩 규칙 최적화"""
|
|
system_prompt = """알고리즘 트레이딩 전략가. JSON으로만 응답.
|
|
필수 필드: strategy_name, market_condition, parameters(buy_conditions,sell_conditions(take_profit_rate,stop_loss_rate,trailing_stop)), risk_assessment, confidence_score(0-100).
|
|
악재시 보수적, 호재시 적극적 설정."""
|
|
|
|
user_message = f"종목:{stock_code}\n시장:{_dumps(market_data)}"
|
|
|
|
if news_data:
|
|
user_message += f"\n뉴스:{_dumps(news_data)}"
|
|
|
|
response = await self.analyze(system_prompt, user_message)
|
|
|
|
# 텍스트 응답 파싱
|
|
for content in response["content"]:
|
|
if content["type"] == "text":
|
|
try:
|
|
text = content["text"]
|
|
json_start = text.find('{')
|
|
json_end = text.rfind('}') + 1
|
|
if json_start != -1 and json_end != -1:
|
|
return json.loads(text[json_start:json_end])
|
|
except Exception as e:
|
|
logger.error(f"Failed to parse optimization result: {e}")
|
|
|
|
return {"error": "Failed to generate optimized rules"}
|
|
|
|
async def generate_trade_journal(
|
|
self,
|
|
trade_info: Dict[str, Any],
|
|
market_data: Dict[str, Any]
|
|
) -> Dict[str, Any]:
|
|
"""매매일지 자동 작성"""
|
|
system_prompt = """매매일지 작성 전문가. JSON으로 응답.
|
|
필수 필드: title, content(간략 Markdown), mood(평정심/흥분/공포/아쉬움/만족), tags(배열), score(0-100).
|
|
매매근거, 잘한점/아쉬운점, 개선사항 포함."""
|
|
|
|
user_message = f"거래:{_dumps(trade_info)}\n시장:{_dumps(market_data)}"
|
|
|
|
response = await self.analyze(system_prompt, user_message)
|
|
|
|
for content in response["content"]:
|
|
if content["type"] == "text":
|
|
try:
|
|
text = content["text"]
|
|
json_start = text.find('{')
|
|
json_end = text.rfind('}') + 1
|
|
if json_start != -1 and json_end != -1:
|
|
return json.loads(text[json_start:json_end])
|
|
except Exception as e:
|
|
logger.error(f"Failed to parse journal result: {e}")
|
|
|
|
return {"error": "Failed to generate journal"}
|
|
|
|
|
|
claude_client = ClaudeClient()
|