diff --git a/backend/alembic.ini b/backend/alembic.ini new file mode 100644 index 0000000..3614cfd --- /dev/null +++ b/backend/alembic.ini @@ -0,0 +1,43 @@ +[alembic] +script_location = alembic +prepend_sys_path = . +version_path_separator = os + +# DATABASE_URL은 env.py 에서 app.core.config.settings 로 읽음 +sqlalchemy.url = + +[post_write_hooks] + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/backend/alembic/env.py b/backend/alembic/env.py new file mode 100644 index 0000000..76ea40f --- /dev/null +++ b/backend/alembic/env.py @@ -0,0 +1,86 @@ +"""Alembic 환경 설정 (async / asyncpg) + +- settings.DATABASE_URL 에서 연결 문자열을 읽어 동기 드라이버(psycopg2) URL 로 변환 +- 모델 메타데이터를 임포트해 autogenerate 지원 +""" +import asyncio +from logging.config import fileConfig + +from sqlalchemy import pool +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import async_engine_from_config + +from alembic import context + +# ── 프로젝트 루트를 sys.path 에 추가 (alembic.ini prepend_sys_path = .) ─────── +import sys +import os + +# backend/ 가 작업 디렉토리이므로 app 패키지를 바로 임포트 가능 +sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + +from app.core.config import settings +from app.core.database import Base + +# 모든 모델 임포트 — autogenerate 가 테이블을 인식하도록 +import app.models.models # noqa: F401 + +# ── alembic.ini 로거 설정 ───────────────────────────────────────────────────── +config = context.config +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +# ── DB URL: asyncpg → psycopg2 (Alembic 동기 드라이버용) ──────────────────── +_db_url = settings.DATABASE_URL +if "+asyncpg" in _db_url: + _db_url = _db_url.replace("+asyncpg", "") +elif _db_url.startswith("postgresql://"): + pass # already sync-compatible +config.set_main_option("sqlalchemy.url", _db_url) + +target_metadata = Base.metadata + + +# ── 오프라인 마이그레이션 ───────────────────────────────────────────────────── +def run_migrations_offline() -> None: + url = config.get_main_option("sqlalchemy.url") + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + with context.begin_transaction(): + context.run_migrations() + + +# ── 온라인 마이그레이션 (async) ─────────────────────────────────────────────── +def do_run_migrations(connection: Connection) -> None: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations() -> None: + _async_url = settings.DATABASE_URL + if not "+asyncpg" in _async_url: + _async_url = _async_url.replace("postgresql://", "postgresql+asyncpg://") + + connectable = async_engine_from_config( + {"sqlalchemy.url": _async_url}, + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + async with connectable.connect() as connection: + await connection.run_sync(do_run_migrations) + await connectable.dispose() + + +def run_migrations_online() -> None: + asyncio.run(run_async_migrations()) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/backend/alembic/script.py.mako b/backend/alembic/script.py.mako new file mode 100644 index 0000000..fbc4b07 --- /dev/null +++ b/backend/alembic/script.py.mako @@ -0,0 +1,26 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[str, None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/backend/alembic/versions/0001_initial_schema_columns.py b/backend/alembic/versions/0001_initial_schema_columns.py new file mode 100644 index 0000000..e23eaa0 --- /dev/null +++ b/backend/alembic/versions/0001_initial_schema_columns.py @@ -0,0 +1,122 @@ +"""initial schema columns + +lifespan 에서 관리하던 ADD COLUMN / CREATE TABLE DDL 을 Alembic 으로 이전. +이 revision 이 적용된 후에는 main.py lifespan 의 raw SQL DDL 블록은 제거됨. + +Revision ID: 0001 +Revises: +Create Date: 2026-02-27 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision: str = "0001" +down_revision: Union[str, None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ── trade_history 컬럼 추가 ───────────────────────────────────────────── + with op.batch_alter_table("trade_history", schema=None) as batch_op: + for col_name, col_type in [ + ("total_amount", sa.Numeric(15, 2)), + ("entry_reason", sa.Text()), + ("exit_reason", sa.Text()), + ("signals", postgresql.JSONB()), + ("entry_trade_id", sa.Integer()), + ("slippage", sa.Numeric(10, 2)), + ("commission", sa.Numeric(10, 2)), + ("net_profit_loss", sa.Numeric(15, 2)), + ("profit_loss_rate", sa.Numeric(10, 4)), + ]: + try: + batch_op.add_column(sa.Column(col_name, col_type, nullable=True)) + except Exception: + pass # 이미 존재하는 컬럼은 무시 + + # total_amount 백필 + op.execute(""" + UPDATE trade_history + SET total_amount = executed_price * executed_quantity + WHERE total_amount IS NULL + AND executed_price IS NOT NULL + AND executed_quantity IS NOT NULL + """) + + # ── stock_analysis 컬럼 추가 ──────────────────────────────────────────── + with op.batch_alter_table("stock_analysis", schema=None) as batch_op: + for col_name, col_type in [ + ("atr_14", sa.Numeric(15, 2)), + ("adx_14", sa.Numeric(10, 2)), + ("ml_score", sa.Integer()), + ("ml_confidence", sa.Numeric(5, 4)), + ("ml_signal", sa.String(10)), + ]: + try: + batch_op.add_column(sa.Column(col_name, col_type, nullable=True)) + except Exception: + pass + + # stock_analysis 유니크 제약 추가 (idempotent) + op.execute(""" + DO $$ BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'uq_stock_analysis_code_date' + AND conrelid = 'stock_analysis'::regclass + ) THEN + ALTER TABLE stock_analysis + ADD CONSTRAINT uq_stock_analysis_code_date + UNIQUE (stock_code, analysis_date); + END IF; + END $$; + """) + + # ── active_trades 컬럼 추가 ───────────────────────────────────────────── + with op.batch_alter_table("active_trades", schema=None) as batch_op: + for col_name, col_type, default in [ + ("remaining_pct", sa.Numeric(5, 2), "1.0"), + ("highest_since_entry", sa.Numeric(15, 2), None), + ("strategy_name", sa.String(100), None), + ]: + try: + col = sa.Column(col_name, col_type, nullable=True, + server_default=default if default else sa.null()) + batch_op.add_column(col) + except Exception: + pass + + # ── trading_sessions 테이블 ───────────────────────────────────────────── + op.execute(""" + CREATE TABLE IF NOT EXISTS trading_sessions ( + id SERIAL PRIMARY KEY, + session_date DATE UNIQUE NOT NULL, + consecutive_losses INTEGER DEFAULT 0, + daily_profit_loss NUMERIC(15, 2) DEFAULT 0, + is_paused BOOLEAN DEFAULT FALSE, + pause_reason TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # ── stock_bans 테이블 ─────────────────────────────────────────────────── + op.execute(""" + CREATE TABLE IF NOT EXISTS stock_bans ( + stock_code VARCHAR(20) PRIMARY KEY, + consecutive_losses INTEGER DEFAULT 0 NOT NULL, + ban_until TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + +def downgrade() -> None: + # downgrade 는 데이터 손실 위험이 있으므로 명시적으로 구현하지 않음 + # 필요 시 ALTER TABLE DROP COLUMN 을 수동으로 실행 + pass diff --git a/backend/app/api/endpoints/day_trading.py b/backend/app/api/endpoints/day_trading.py index 3f37115..15346bf 100644 --- a/backend/app/api/endpoints/day_trading.py +++ b/backend/app/api/endpoints/day_trading.py @@ -101,10 +101,17 @@ _last_reset_date: Optional[date] = None # 매매 실행 lock — _positions/_daily_pnl/_closed_today 변경 시 반드시 획득 _trade_lock = asyncio.Lock() +# 포트폴리오 요약 캐시 (get_balance() 호출 최소화) +_portfolio_summary_cache: Dict[str, float] = {} +_portfolio_summary_ts: Optional[datetime] = None +_portfolio_cache_dirty: bool = True # 최초 또는 매매 발생 시 강제 갱신 +_PORTFOLIO_CACHE_TTL: float = 300.0 # 매매 없을 때 최대 5분 캐시 유지 + def _reset_if_new_day() -> None: """날짜가 바뀌면 일일 상태 초기화""" global _daily_pnl, _orb_cache, _last_reset_date, _last_total_assets, _closed_today + global _portfolio_summary_cache, _portfolio_summary_ts, _portfolio_cache_dirty today = _today_kst() if _last_reset_date != today: _daily_pnl = 0.0 @@ -112,6 +119,9 @@ def _reset_if_new_day() -> None: _last_total_assets = 0.0 _closed_today = set() _last_reset_date = today + _portfolio_summary_cache = {} + _portfolio_summary_ts = None + _portfolio_cache_dirty = True logger.info(f"📅 데이트레이딩 상태 일일 초기화: {today}") @@ -143,18 +153,39 @@ async def _get_current_price(stock_code: str) -> float: return 0.0 -async def _get_portfolio_summary() -> Dict[str, float]: - """포트폴리오 요약 (총자산, 가용 현금)""" +async def _get_portfolio_summary(force: bool = False) -> Dict[str, float]: + """포트폴리오 요약 (총자산, 가용 현금). + + 매매가 발생했거나 캐시 TTL이 만료된 경우에만 KIS get_balance() 호출. + 그 외에는 인메모리 캐시 반환 (API 호출 최소화). + """ + global _portfolio_summary_cache, _portfolio_summary_ts, _portfolio_cache_dirty + + now = _kst_now() + cache_age = (now - _portfolio_summary_ts).total_seconds() if _portfolio_summary_ts else 9999.0 + cache_valid = ( + not _portfolio_cache_dirty + and cache_age < _PORTFOLIO_CACHE_TTL + and bool(_portfolio_summary_cache) + ) + + if not force and cache_valid: + logger.debug(f"⏭️ Portfolio summary cache hit (age={cache_age:.0f}s) — skipping get_balance()") + return _portfolio_summary_cache + try: balance = await korea_investment_api.get_balance() output2 = balance.get("output2", [{}]) summary = output2[0] if output2 else {} total_assets = float(summary.get("tot_evlu_amt", 0) or 0) available_cash = float(summary.get("prvs_rcdl_excc_amt", 0) or 0) - return {"total_assets": total_assets, "available_cash": available_cash} + _portfolio_summary_cache = {"total_assets": total_assets, "available_cash": available_cash} + _portfolio_summary_ts = now + _portfolio_cache_dirty = False + return _portfolio_summary_cache except Exception as e: logger.error(f"포트폴리오 조회 실패: {e}") - return {"total_assets": 0.0, "available_cash": 0.0} + return _portfolio_summary_cache or {"total_assets": 0.0, "available_cash": 0.0} async def _place_buy_order( @@ -412,6 +443,7 @@ async def _check_exits( ok = await _place_sell_order(code, quantity, price=0) # 시장가 if ok: pnl = (current_price - pos["entry_price"]) * quantity + _portfolio_cache_dirty = True # 매도 체결 → 잔고 변경됨 async with _trade_lock: _daily_pnl += pnl _positions.pop(code, None) @@ -580,6 +612,9 @@ async def _execute_entry(signal: DayTradeSignal, quantity: int) -> Optional[Dict if not order_no: return None + global _portfolio_cache_dirty + _portfolio_cache_dirty = True # 매수 체결 → 잔고 변경됨 + _positions[signal.stock_code] = { "stock_name": signal.stock_name, "strategy": signal.strategy, @@ -643,6 +678,7 @@ async def _execute_eod_flat() -> None: ok = await _place_sell_order(code, pos["quantity"], price=0) if ok: pnl = (current_price - pos["entry_price"]) * pos["quantity"] + _portfolio_cache_dirty = True # EOD 청산 → 잔고 변경됨 async with _trade_lock: _daily_pnl += pnl _positions.pop(code, None) diff --git a/backend/app/api/endpoints/portfolio.py b/backend/app/api/endpoints/portfolio.py index 2556d9f..79913ac 100644 --- a/backend/app/api/endpoints/portfolio.py +++ b/backend/app/api/endpoints/portfolio.py @@ -606,7 +606,13 @@ async def check_and_execute_exit_trades(db: AsyncSession): from datetime import datetime try: - await sync_portfolio(db, force=True) + # 매매가 발생했거나 캐시가 비어있을 때만 KIS 잔고 조회 (API 호출 최소화) + latest_trade_id = await _get_latest_executed_trade_id(db) + if latest_trade_id != portfolio_cache.last_trade_id or not portfolio_cache.holdings: + logger.info(f"🔄 New trade detected (prev={portfolio_cache.last_trade_id}, cur={latest_trade_id}) — syncing balance") + await sync_portfolio(db, force=True) + else: + logger.debug("⏭️ No new trade since last sync — skipping get_balance()") # MDD 업데이트 (async — thread-safe lock 내부) summary = await _get_portfolio_summary(db) diff --git a/backend/app/main.py b/backend/app/main.py index 4241ec5..c7a85b6 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -23,7 +23,6 @@ from app.services.db_worker import db_writer_worker from app.core.database import engine, Base, AsyncSessionLocal from app.models import models # 모델 임포트 from app.core.market_schedule import is_market_open -from sqlalchemy import text # 로깅 설정 import os @@ -267,8 +266,9 @@ async def auto_trade_scheduler(): try: from app.services.db_worker import enqueue_db_write await enqueue_db_write({ - "action": "log_mcp_deny", + "type": "log_mcp_deny", "reason": reason, + "trade_plan": trade_plan, }) except Exception: pass @@ -338,77 +338,14 @@ async def lifespan(app: FastAPI): logger.info(f"KIS Account: {settings.KOREA_INVESTMENT_ACCOUNT_NUMBER}") # DB 테이블 자동 생성 + # 컬럼 추가/테이블 생성 DDL 은 Alembic migration 으로 관리됨. + # 배포 전 `cd backend && alembic upgrade head` 를 실행할 것. + # (alembic/versions/0001_initial_schema_columns.py) try: logger.info(f"Connecting to database: {settings.DATABASE_URL.split('@')[1] if '@' in settings.DATABASE_URL else 'unknown'}") async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) - await conn.execute(text(""" - ALTER TABLE trade_history - ADD COLUMN IF NOT EXISTS total_amount NUMERIC(15, 2), - ADD COLUMN IF NOT EXISTS entry_reason TEXT, - ADD COLUMN IF NOT EXISTS exit_reason TEXT, - ADD COLUMN IF NOT EXISTS signals JSONB, - ADD COLUMN IF NOT EXISTS entry_trade_id INTEGER, - ADD COLUMN IF NOT EXISTS slippage NUMERIC(10, 2), - ADD COLUMN IF NOT EXISTS commission NUMERIC(10, 2), - ADD COLUMN IF NOT EXISTS net_profit_loss NUMERIC(15, 2), - ADD COLUMN IF NOT EXISTS profit_loss_rate NUMERIC(10, 4); - """)) - await conn.execute(text(""" - UPDATE trade_history - SET total_amount = executed_price * executed_quantity - WHERE total_amount IS NULL - AND executed_price IS NOT NULL - AND executed_quantity IS NOT NULL; - """)) - await conn.execute(text(""" - ALTER TABLE stock_analysis - ADD COLUMN IF NOT EXISTS atr_14 NUMERIC(15, 2), - ADD COLUMN IF NOT EXISTS adx_14 NUMERIC(10, 2), - ADD COLUMN IF NOT EXISTS ml_score INTEGER, - ADD COLUMN IF NOT EXISTS ml_confidence NUMERIC(5, 4), - ADD COLUMN IF NOT EXISTS ml_signal VARCHAR(10); - """)) - await conn.execute(text(""" - DO $$ BEGIN - IF NOT EXISTS ( - SELECT 1 FROM pg_constraint - WHERE conname = 'uq_stock_analysis_code_date' - AND conrelid = 'stock_analysis'::regclass - ) THEN - ALTER TABLE stock_analysis - ADD CONSTRAINT uq_stock_analysis_code_date - UNIQUE (stock_code, analysis_date); - END IF; - END $$; - """)) - await conn.execute(text(""" - ALTER TABLE active_trades - ADD COLUMN IF NOT EXISTS remaining_pct NUMERIC(5, 2) DEFAULT 1.0, - ADD COLUMN IF NOT EXISTS highest_since_entry NUMERIC(15, 2), - ADD COLUMN IF NOT EXISTS strategy_name VARCHAR(100); - """)) - await conn.execute(text(""" - CREATE TABLE IF NOT EXISTS trading_sessions ( - id SERIAL PRIMARY KEY, - session_date DATE UNIQUE NOT NULL, - consecutive_losses INTEGER DEFAULT 0, - daily_profit_loss NUMERIC(15, 2) DEFAULT 0, - is_paused BOOLEAN DEFAULT FALSE, - pause_reason TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ); - """)) - await conn.execute(text(""" - CREATE TABLE IF NOT EXISTS stock_bans ( - stock_code VARCHAR(20) PRIMARY KEY, - consecutive_losses INTEGER DEFAULT 0 NOT NULL, - ban_until TIMESTAMP, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ); - """)) - logger.info("✅ Database tables created successfully (and columns verified)") + logger.info("✅ Database tables created successfully") try: from app.services.loss_tracker import loss_tracker @@ -556,8 +493,16 @@ app.add_middleware( logger.info("CORS middleware configured") +_LOG_SKIP_PREFIXES = ("/health", "/ws/") + + @app.middleware("http") async def log_requests(request: Request, call_next): + path = request.url.path + # /health, /ws/ 는 폴링이 잦아 로그 노이즈가 크므로 debug 레벨로 낮춤 + if any(path.startswith(p) for p in _LOG_SKIP_PREFIXES): + return await call_next(request) + start_time = time.time() logger.info(f"🔵 Request: {request.method} {request.url}") response = await call_next(request) diff --git a/backend/app/services/db_worker.py b/backend/app/services/db_worker.py index 7f58c01..ef3d705 100644 --- a/backend/app/services/db_worker.py +++ b/backend/app/services/db_worker.py @@ -9,6 +9,7 @@ Supported task types: - insert_trade_history : dict → TradeHistory INSERT - upsert_stock_analysis : dict → StockAnalysis single UPSERT - bulk_upsert_stock_analysis : list → StockAnalysis batch UPSERT + - log_mcp_deny : dict → MCP 거부 이벤트 로그 (DB 저장 없이 로그만) """ import asyncio import logging @@ -44,6 +45,15 @@ async def enqueue_bulk_upsert_analysis(rows: list[dict]) -> None: await db_task_queue.put({"type": "bulk_upsert_stock_analysis", "data": rows}) +async def enqueue_db_write(task: dict) -> None: + """범용 DB 쓰기 태스크를 큐에 등록. + + task 는 반드시 'type' 키를 포함해야 하며, 워커가 지원하는 task type 중 하나여야 한다. + 예: {"type": "log_mcp_deny", "reason": "...", "trade_plan": {...}} + """ + await db_task_queue.put(task) + + # ───────────────────────────────────────────────────────────────────────────── # 워커 본체 # ───────────────────────────────────────────────────────────────────────────── @@ -89,6 +99,15 @@ async def db_writer_worker() -> None: "✅ [Worker] Bulk StockAnalysis upsert: %d rows", len(data) ) + # ── 4. MCP 거부 이벤트 로그 ─────────────────────────────────── + elif task_type == "log_mcp_deny": + # DB 저장 없이 로그만 기록 (필요 시 별도 테이블 확장 가능) + logger.warning( + "🛑 [Worker] MCP DENY logged: reason=%s, trade_plan=%s", + task.get("reason", "unknown"), + task.get("trade_plan"), + ) + else: logger.warning("⚠️ [Worker] Unknown task type: %s", task_type) diff --git a/backend/app/services/market_analyzer.py b/backend/app/services/market_analyzer.py index e1a2d36..f107727 100644 --- a/backend/app/services/market_analyzer.py +++ b/backend/app/services/market_analyzer.py @@ -173,6 +173,10 @@ async def analyze_stock(stock_code: str, stock_name: str, criteria: Recommendati if current_price == 0: return None + # 가격 필터: 5,000원 미만 종목 제외 (CLAUDE.md 섹션 5.4) + if current_price < settings.MIN_PRICE_KRW: + return None + # 전일 거래량 필터: 100만주 미만 종목 제외 (유동성 부족) if volume < 1_000_000: return None diff --git a/backend/app/services/trade_executor.py b/backend/app/services/trade_executor.py index 1abd681..661cd1b 100644 --- a/backend/app/services/trade_executor.py +++ b/backend/app/services/trade_executor.py @@ -20,6 +20,9 @@ from app.services.technical_indicators import ( logger = logging.getLogger(__name__) +# 동시 주문 1개로 제한 — 중복 주문 및 레이스 컨디션 방지 +_order_semaphore = asyncio.Semaphore(1) + async def _mcp_validate_buy(stock: dict) -> tuple: if not settings.MCP_VALIDATION_ENABLED: @@ -63,10 +66,9 @@ async def _mcp_validate_buy(stock: dict) -> tuple: return True, f"mcp_approved(conf={confidence:.2f})" else: return False, f"mcp_rejected(decision={decision}, conf={confidence:.2f})" - + except httpx.HTTPStatusError as e: if e.response.status_code == 500: - # 500 에러 → MCP 서버 내부 오류 → 룰 기반으로 우회 logger.warning(f"⚠️ MCP 서버 500 에러 — 룰 기반으로 우회: {e}") return True, "mcp_500_fallback_to_rules" logger.warning(f"⚠️ MCP 검증 실패 — 매매 차단: {e}") @@ -77,15 +79,319 @@ async def _mcp_validate_buy(stock: dict) -> tuple: return False, "mcp_unavailable" +async def _execute_buy( + stock: dict, + db: AsyncSession, + holding, + current_price: float, + signals: list, + score_val: float, + is_domestic: bool, + exchange_code: str, + stock_code: str, + stock_name: str, +) -> dict: + """BUY 주문 실행 로직 — 선체크 → semaphore → 주문 → TradeHistory 기록 → sync.""" + + # ── 1. 3일 급등 체크 ───────────────────────────────────────────────────── + try: + if is_domestic: + chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D", limit=10) + chart_data = chart_resp.get("output2", []) + else: + chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") + chart_data = chart_resp.get("output2", []) + + if len(chart_data) >= 4: + closes = ( + [float(item.get("stck_clpr", 0)) for item in chart_data[-4:]] + if is_domestic + else [float(item.get("clos", 0)) for item in chart_data[-4:]] + ) + if len(closes) >= 4: + price_3days_ago = closes[0] + recent_gain = ( + ((current_price - price_3days_ago) / price_3days_ago) * 100 + if price_3days_ago > 0 else 0 + ) + if recent_gain > 10: + return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"} + except Exception as e: + logger.debug(f"[chase_buy] 차트 조회 실패, 우회: {stock_code} — {e}") + + # ── 2. 트렌드 필터 ──────────────────────────────────────────────────────── + if TREND_FILTER_ENABLED: + try: + ma20 = stock.get("ma20") + ma60 = stock.get("ma60") + if ma20 is None or ma60 is None: + if is_domestic: + chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D") + closes_local = [float(it.get("stck_clpr", 0)) for it in (chart_resp.get("output") or [])] + else: + chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") + closes_local = [float(it.get("clos", 0)) for it in (chart_resp.get("output2") or [])] + + if len(closes_local) >= max(TREND_MA_SHORT, TREND_MA_LONG): + ma20 = _ma(closes_local, TREND_MA_SHORT) + ma60 = _ma(closes_local, TREND_MA_LONG) + + if not ( + ma20 and ma60 + and float(ma20) >= float(ma60) * (1 - TREND_TOLERANCE_PCT) + and current_price >= float(ma20) * (1 - TREND_TOLERANCE_PCT) + ): + return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter"} + except Exception: + return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter_error"} + + # ── 3. 현금·자산 조회 ──────────────────────────────────────────────────── + try: + from app.api.endpoints.portfolio import portfolio_cache + available_cash = float(portfolio_cache.summary.get("available_cash", 0)) if portfolio_cache.summary else 0.0 + total_assets = float(portfolio_cache.summary.get("total_assets", 0)) if portfolio_cache.summary else 0.0 + except Exception: + available_cash, total_assets = 0.0, 0.0 + + daily_cutoff = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) + + # ── 4. PENDING 포함 오늘 중복 주문 차단 ────────────────────────────────── + pending_check = await db.execute( + select(TradeHistory).where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "BUY", + TradeHistory.status.in_(["EXECUTED", "COMPLETED", "PENDING"]), + TradeHistory.created_at >= daily_cutoff, + ) + ) + if pending_check.scalars().first(): + return {"action": "SKIP", "stock": stock_name, "reason": "already_ordered_today"} + + # ── 5. 일 매수 횟수 제한 ───────────────────────────────────────────────── + daily_buy_check = await db.execute( + select(TradeHistory).where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "BUY", + TradeHistory.status.in_(["EXECUTED", "COMPLETED"]), + TradeHistory.created_at >= daily_cutoff, + ) + ) + if len(daily_buy_check.scalars().all()) >= 2: + return {"action": "HOLD", "stock": stock_name, "reason": "daily_trade_limit"} + + # ── 6. 쿨다운 체크 ──────────────────────────────────────────────────────── + cooldown_min = int(stock.get("add_buy_cooldown_min", 30) or 30) + cooldown_check = await db.execute( + select(TradeHistory) + .where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "BUY", + TradeHistory.status.in_(["EXECUTED", "COMPLETED"]), + TradeHistory.created_at >= daily_cutoff, + ) + .order_by(desc(TradeHistory.created_at)) + ) + recent_buys = cooldown_check.scalars().all() + if recent_buys and (_get_kst_now() - recent_buys[0].created_at).total_seconds() < cooldown_min * 60: + return {"action": "HOLD", "stock": stock_name, "reason": "cooldown"} + + # ── 7. MCP 검증 ─────────────────────────────────────────────────────────── + mcp_ok, mcp_reason = await _mcp_validate_buy(stock) + if not mcp_ok: + return {"action": "HOLD", "stock": stock_name, "reason": mcp_reason} + + # ── 8. 포지션 사이징 ────────────────────────────────────────────────────── + current_value = float(holding.quantity * current_price) if holding else 0.0 + regime_str = stock.get("_regime_str", "bear") + regime = MarketRegime.BULL if regime_str == "bull" else MarketRegime.BEAR + invest_amount = calculate_position_size_v2( + total_assets=total_assets, + available_cash=available_cash, + current_strategy_allocation=current_value, + regime=regime, + rsi=stock.get("rsi"), + adx=stock.get("adx_14"), + ) + if invest_amount <= 0 and total_assets <= 0: + invest_amount = available_cash * 0.04 + + quantity = int(invest_amount / current_price) + if quantity <= 0: + return {"action": "SKIP", "reason": "quantity_zero", "stock": stock_name} + + # ── 9. 주문 실행 (semaphore로 동시성 제어) ─────────────────────────────── + try: + async with _order_semaphore: + if is_domestic: + order_result = await korea_investment_api.place_order( + stock_code, "BUY", quantity, int(current_price) + ) + else: + order_result = await korea_investment_api.place_overseas_order( + stock_code, "BUY", quantity, float(current_price), exchange_code + ) + + total_amount = Decimal(str(current_price)) * quantity + entry_reason = f"매수 신호: {', '.join(signals[:3])}. 점수: {score_val}점" + success = order_result.get("rt_cd") == "0" + status = "COMPLETED" if success else ("PENDING" if order_result.get("rt_cd") == "1" else "FAILED") + + trade = TradeHistory( + rule_id=stock.get("rule_id"), + stock_code=stock_code, + stock_name=stock_name, + order_type="BUY", + order_price=Decimal(str(current_price)), + order_quantity=quantity, + executed_price=Decimal(str(current_price)) if success else None, + executed_quantity=quantity if success else 0, + total_amount=total_amount, + status=status, + order_id=order_result.get("odno"), + message=f"자동매수: {'추가매수' if holding else '신규'} | {' | '.join(signals[:2])}", + entry_reason=entry_reason, + signals=signals, + created_at=_get_kst_now(), + executed_at=_get_kst_now() if success else None, + ) + db.add(trade) + await db.commit() + + if success: + # 즉시 캐시 반영 (UI 실시간성) + try: + from app.api.endpoints.portfolio import portfolio_cache + portfolio_cache.apply_buy(stock_code, stock_name, quantity, current_price) + except Exception: + pass + # KIS 체결 확인 후 포트폴리오 DB 동기화 (3초 딜레이) + asyncio.create_task(_delayed_sync_portfolio(delay_seconds=3)) + + return { + "action": "BUY", + "stock": stock_name, + "quantity": quantity, + "price": current_price, + "success": success, + } + + except Exception as e: + return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)} + + +async def _execute_sell( + stock: dict, + db: AsyncSession, + holding, + current_price: float, + signals: list, + is_domestic: bool, + exchange_code: str, + stock_code: str, + stock_name: str, +) -> dict: + """SELL 주문 실행 로직 — 중복체크 선행 → semaphore → 주문 → TradeHistory 기록 → sync.""" + + quantity = holding.quantity + _avg = float(holding.average_price or 0) + profit_loss_rate = ( + ((current_price - _avg) / _avg) * 100 if _avg > 0 + else float(holding.profit_loss_rate or 0) + ) + take_profit = stock.get("take_profit", 3.0) + stop_loss = stock.get("stop_loss", -1.5) + + if not (profit_loss_rate >= take_profit or profit_loss_rate <= stop_loss or True): + return {"action": "HOLD", "stock": stock_name} + + try: + # ── 1. 중복 주문 체크 — 주문 전에 먼저 확인 ──────────────────────── + recent_cutoff = _get_kst_now() - timedelta(seconds=10) + duplicate_check = await db.execute( + select(TradeHistory).where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "SELL", + TradeHistory.created_at >= recent_cutoff, + ) + ) + if duplicate_check.scalar_one_or_none(): + return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name} + + # ── 2. 주문 실행 (semaphore로 동시성 제어) ────────────────────────── + async with _order_semaphore: + if is_domestic: + order_result = await korea_investment_api.place_order( + stock_code, "SELL", quantity, int(current_price) + ) + else: + order_result = await korea_investment_api.place_overseas_order( + stock_code, "SELL", quantity, float(current_price), exchange_code + ) + + total_amount = Decimal(str(current_price)) * quantity + profit_loss = (current_price - float(holding.average_price or current_price)) * quantity + success = order_result.get("rt_cd") == "0" + status = "COMPLETED" if success else ("PENDING" if order_result.get("rt_cd") == "1" else "FAILED") + + trade = TradeHistory( + rule_id=stock.get("rule_id"), + stock_code=stock_code, + stock_name=stock_name, + order_type="SELL", + order_price=Decimal(str(current_price)), + order_quantity=quantity, + executed_price=Decimal(str(current_price)) if success else None, + executed_quantity=quantity if success else 0, + total_amount=total_amount, + status=status, + order_id=order_result.get("odno"), + message=f"자동매도: 손익률 {profit_loss_rate:.2f}%", + exit_reason=f"손익: {profit_loss:,.0f}원", + signals=signals, + net_profit_loss=Decimal(str(profit_loss)) if success else None, + profit_loss_rate=Decimal(str(profit_loss_rate)) if success else None, + created_at=_get_kst_now(), + executed_at=_get_kst_now() if success else None, + ) + db.add(trade) + await db.commit() + + if success: + # 즉시 캐시 반영 + try: + from app.api.endpoints.portfolio import portfolio_cache + avg = float(holding.average_price or current_price) + portfolio_cache.apply_sell(stock_code, quantity, current_price, avg) + portfolio_cache.holdings = [ + h for h in portfolio_cache.holdings if h.stock_code != stock_code + ] + except Exception: + pass + # KIS 체결 확인 후 포트폴리오 DB 동기화 + asyncio.create_task(_delayed_sync_portfolio(delay_seconds=3)) + + return { + "action": "SELL", + "stock": stock_name, + "quantity": quantity, + "price": current_price, + "profit_loss": profit_loss_rate, + "success": success, + } + + except Exception as e: + return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)} + + async def execute_auto_trade_background(recommendations_response: dict, db: AsyncSession): - # recommendations_response should be a dict containing 'recommendations_by_rule'. + """recommendations_response dict 에서 추천 종목을 꺼내 자동매매 실행.""" if not isinstance(recommendations_response, dict): logger.error("recommendations_response is not a dict: %r", recommendations_response) return try: - trade_results = [] total_processed = 0 + trade_results = [] market_regime = None if settings.BEAR_MARKET_FILTER_ENABLED: @@ -94,14 +400,12 @@ async def execute_auto_trade_background(recommendations_response: dict, db: Asyn regime_max_trades = market_regime.get("max_trades", 20) if market_regime else 20 position_multiplier = market_regime.get("position_multiplier", 1.0) if market_regime else 1.0 max_trades = min(20, regime_max_trades) - + result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) held_stocks = {p.stock_code for p in result.scalars().all()} - - # normalize recommendations_by_rule early so we iterate over a list of groups + rec_groups = recommendations_response.get("recommendations_by_rule", []) if isinstance(rec_groups, dict): - # older format: {rule_name: [stocks]} normalized_list = [] for name, group in rec_groups.items(): if isinstance(group, dict): @@ -109,58 +413,57 @@ async def execute_auto_trade_background(recommendations_response: dict, db: Asyn else: normalized_list.append({"rule_name": name, "stocks": group}) rec_groups = normalized_list - - unique_stocks = {} + + unique_stocks: dict = {} for rule_group in rec_groups: - # make sure we have a dict with expected keys if not isinstance(rule_group, dict): - logger.warning("Invalid rule_group in recommendations_response: %r", rule_group) + logger.warning("Invalid rule_group: %r", rule_group) continue stocks_list = rule_group.get("stocks") if not isinstance(stocks_list, list): - logger.warning("Invalid stocks list in rule_group: %r", stocks_list) + logger.warning("Invalid stocks list: %r", stocks_list) continue - for stock in stocks_list: if not isinstance(stock, dict): logger.warning("Skipping non-dict stock entry: %r", stock) continue - stock_code = stock.get("stock_code") if stock_code not in unique_stocks: unique_stocks[stock_code] = stock elif stock.get("rule_priority", 999) < unique_stocks[stock_code].get("rule_priority", 999): unique_stocks[stock_code] = stock - + sorted_stocks = sorted( unique_stocks.values(), - key=lambda s: (s.get("rule_priority", 999), -s.get("score", 0)) + key=lambda s: (s.get("rule_priority", 999), -s.get("score", 0)), ) - - for stock in sorted_stocks[:max_trades]: - # ensure stock is a dict before proceeding - if not isinstance(stock, dict): - logger.warning("Skipping invalid stock entry in sorted_stocks: %r", stock) - continue + for stock in sorted_stocks[:max_trades]: + if not isinstance(stock, dict): + logger.warning("Skipping invalid stock entry: %r", stock) + continue try: stock["_position_multiplier"] = position_multiplier - stock["_regime_str"] = market_regime.get("regime_200d", "bear") if market_regime else "bear" + stock["_regime_str"] = ( + market_regime.get("regime_200d", "bear") if market_regime else "bear" + ) result = await execute_auto_trade(stock, db) total_processed += 1 - + if result["action"] in ["BUY", "SELL"]: trade_results.append(result) - + await asyncio.sleep(0.5) - + except Exception as e: - # stock is dict so we can safely access get logger.error(f"Error executing auto-trade for {stock.get('stock_code')}: {e}") continue - + except Exception as e: - logger.error(f"❌ Background auto-trade error: {e} - recommendations_response={recommendations_response!r}") + logger.error( + f"❌ Background auto-trade error: {e} — " + f"recommendations_response={recommendations_response!r}" + ) async def _background_sync_portfolio(): @@ -176,7 +479,7 @@ async def _delayed_sync_portfolio(delay_seconds: int = 3): try: await asyncio.sleep(delay_seconds) from app.api.endpoints.portfolio import sync_portfolio - + async with AsyncSessionLocal() as session: try: await sync_portfolio(session, force=True) @@ -187,6 +490,11 @@ async def _delayed_sync_portfolio(delay_seconds: int = 3): async def execute_auto_trade(stock: dict, db: AsyncSession): + """종목 단건 매매 판단 및 실행. + + 공통 필터(가격·MDD·손실추적·보유수·당일매도쿨다운·신호분석)를 통과한 후 + _execute_buy / _execute_sell 에 위임한다. + """ try: stock_code = stock.get("stock_code") stock_name = stock.get("stock_name") @@ -199,6 +507,7 @@ async def execute_auto_trade(stock: dict, db: AsyncSession): is_domestic = stock_code.isdigit() and len(stock_code) == 6 exchange_code = stock.get("exchange_code", "NASD") + # ── 가격 필터 ───────────────────────────────────────────────────────── if is_domestic: if current_price < settings.MIN_PRICE_KRW: return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name} @@ -206,6 +515,7 @@ async def execute_auto_trade(stock: dict, db: AsyncSession): if current_price < settings.MIN_PRICE_USD: return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name} + # ── MDD / 손실추적 / 보유수 ────────────────────────────────────────── if not await mdd_manager.can_open_new_position(): return {"action": "SKIP", "reason": "mdd_pause", "stock": stock_name} @@ -213,30 +523,32 @@ async def execute_auto_trade(stock: dict, db: AsyncSession): return {"action": "SKIP", "reason": "consecutive_loss_ban", "stock": stock_name} held_result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) - held_count = len(held_result.scalars().all()) - if held_count >= settings.MAX_HOLDINGS_COUNT: + if len(held_result.scalars().all()) >= settings.MAX_HOLDINGS_COUNT: return {"action": "SKIP", "reason": "max_holdings_reached", "stock": stock_name} + # ── 당일 매도 후 재매수 쿨다운 ─────────────────────────────────────── cutoff_time = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) recent_sell = await db.execute( select(TradeHistory).where( TradeHistory.stock_code == stock_code, TradeHistory.order_type == "SELL", TradeHistory.status == "EXECUTED", - TradeHistory.executed_at >= cutoff_time + TradeHistory.executed_at >= cutoff_time, ) ) if recent_sell.first(): return {"action": "SKIP", "reason": "cooldown_sell_today", "stock": stock_name} + # ── 현재 보유 조회 ──────────────────────────────────────────────────── result = await db.execute( select(Portfolio).where( Portfolio.stock_code == stock_code, - Portfolio.quantity > 0 + Portfolio.quantity > 0, ) ) holding = result.scalar_one_or_none() + # ── 신호 분석 ───────────────────────────────────────────────────────── text_signals = " ".join(signals) try: score_val = float(stock.get("score", 0) or 0) @@ -246,19 +558,17 @@ async def execute_auto_trade(stock: dict, db: AsyncSession): buy_keywords = [ "정배열 눌림목 매수", "눌림목 진입 매수타점", "RSI 바닥 탈출 반등", "RSI 안정 상승", "스토캐스틱 과매도", "5일선 > 20일선", "20일선 > 60일선", - "골든크로스", "볼린저 하한선 터치", "횡보장 — 평균회귀 유리", "거래량 증가와 상승" + "골든크로스", "볼린저 하한선 터치", "횡보장 — 평균회귀 유리", "거래량 증가와 상승", ] - # 거래량 부족 신호만 있는 경우 매수 금지 — 유효 신호에서 거래량부족 제거 후 판단 signals_without_volume_weak = [s for s in signals if "거래량부족" not in s] text_signals_filtered = " ".join(signals_without_volume_weak) should_buy = any(k in text_signals_filtered for k in buy_keywords) - # 거래량부족 신호만 있고 다른 유효 신호가 없으면 매수 금지 if not signals_without_volume_weak and signals: should_buy = False sell_keywords = [ "역배열 하락 매도타점", "RSI 과매수", "스토캐스틱 과매수", "데드크로스", - "이동평균 역배열", "볼린저 상한선 돌파", "손절라인 이탈" + "이동평균 역배열", "볼린저 상한선 돌파", "손절라인 이탈", ] should_sell = any(k in text_signals for k in sell_keywords) @@ -267,7 +577,11 @@ async def execute_auto_trade(stock: dict, db: AsyncSession): ml_signal_trade = stock.get("ml_signal") if settings.ML_ENSEMBLE_ENABLED and ml_score_trade is not None: - if ml_signal_trade == "BUY" and ml_score_trade >= settings.ML_ENSEMBLE_BUY_THRESHOLD and (ml_confidence_trade or 0) >= settings.ML_ENSEMBLE_CONFIDENCE_MIN: + if ( + ml_signal_trade == "BUY" + and ml_score_trade >= settings.ML_ENSEMBLE_BUY_THRESHOLD + and (ml_confidence_trade or 0) >= settings.ML_ENSEMBLE_CONFIDENCE_MIN + ): should_buy = True elif ml_signal_trade == "SELL" and (ml_confidence_trade or 0) >= 0.5: should_sell = True @@ -284,259 +598,34 @@ async def execute_auto_trade(stock: dict, db: AsyncSession): ml_hold_active = settings.ML_ENSEMBLE_ENABLED and ml_signal_trade == "HOLD" and not should_buy if not ml_hold_active: should_buy = True - elif not has_rule: - # rule 없는 경우 급등과열 신호도 차단 + else: if "급등과열" in text_signals or "이격도 과열" in text_signals or "매수금지" in text_signals: return {"action": "HOLD", "stock": stock_name, "reason": "overheat_no_rule"} if should_buy and not has_rule: - strong_buy_kw = ["RSI 바닥 탈출 반등", "볼린저 하한선 터치", "눌림목 진입 매수타점", "정배열 눌림목 매수"] + strong_buy_kw = [ + "RSI 바닥 탈출 반등", "볼린저 하한선 터치", + "눌림목 진입 매수타점", "정배열 눌림목 매수", + ] has_strong_signal = any(k in text_signals_filtered for k in strong_buy_kw) matched_count = sum(1 for k in buy_keywords if k in text_signals_filtered) if not has_strong_signal and (matched_count < 2 or score_val < 2): should_buy = False + # ── 매매 실행 위임 ──────────────────────────────────────────────────── if should_buy: - try: - if is_domestic: - chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D", limit=10) - chart_data = chart_resp.get("output2", []) - else: - chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") - chart_data = chart_resp.get("output2", []) - - if len(chart_data) >= 4: - if is_domestic: - closes = [float(item.get("stck_clpr", 0)) for item in chart_data[-4:]] - else: - closes = [float(item.get("clos", 0)) for item in chart_data[-4:]] - - if len(closes) >= 4: - price_3days_ago = closes[0] - recent_gain = ((current_price - price_3days_ago) / price_3days_ago) * 100 if price_3days_ago > 0 else 0 - if recent_gain > 10: - return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"} - except Exception as e: - pass - - if TREND_FILTER_ENABLED: - try: - ma20 = stock.get("ma20") - ma60 = stock.get("ma60") - if ma20 is None or ma60 is None: - if is_domestic: - chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D") - closes_local = [float(it.get("stck_clpr", 0)) for it in (chart_resp.get("output") or [])] - else: - chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") - closes_local = [float(it.get("clos", 0)) for it in (chart_resp.get("output2") or [])] - - if len(closes_local) >= max(TREND_MA_SHORT, TREND_MA_LONG): - ma20 = _ma(closes_local, TREND_MA_SHORT) - ma60 = _ma(closes_local, TREND_MA_LONG) - if not (ma20 and ma60 and (float(ma20) >= float(ma60) * (1 - TREND_TOLERANCE_PCT)) and (current_price >= float(ma20) * (1 - TREND_TOLERANCE_PCT))): - return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter"} - except Exception as e: - return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter_error"} - - try: - from app.api.endpoints.portfolio import portfolio_cache - available_cash = float(portfolio_cache.summary.get("available_cash", 0)) if portfolio_cache.summary else 0.0 - total_assets = float(portfolio_cache.summary.get("total_assets", 0)) if portfolio_cache.summary else 0.0 - except Exception: - available_cash, total_assets = 0.0, 0.0 - - daily_cutoff = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) - - # PENDING 포함 오늘 전체 중복 주문 차단 (가장 중요한 체크 — 맨 먼저 실행) - pending_check = await db.execute( - select(TradeHistory).where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "BUY", - TradeHistory.status.in_(['EXECUTED', 'COMPLETED', 'PENDING']), - TradeHistory.created_at >= daily_cutoff, - ) + return await _execute_buy( + stock, db, holding, current_price, signals, score_val, + is_domestic, exchange_code, stock_code, stock_name, ) - pending_orders = pending_check.scalars().all() - if pending_orders: - return {"action": "SKIP", "stock": stock_name, "reason": "already_ordered_today"} - daily_buy_check = await db.execute( - select(TradeHistory) - .where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "BUY", - TradeHistory.status.in_(['EXECUTED', 'COMPLETED']), - TradeHistory.created_at >= daily_cutoff, - ) + if should_sell and holding: + return await _execute_sell( + stock, db, holding, current_price, signals, + is_domestic, exchange_code, stock_code, stock_name, ) - daily_buys = daily_buy_check.scalars().all() - if len(daily_buys) >= 2: - return {"action": "HOLD", "stock": stock_name, "reason": "daily_trade_limit"} - cooldown_min = int(stock.get("add_buy_cooldown_min", 30) or 30) - - cooldown_check = await db.execute( - select(TradeHistory) - .where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "BUY", - TradeHistory.status.in_(['EXECUTED', 'COMPLETED']), - TradeHistory.created_at >= daily_cutoff, - ) - .order_by(desc(TradeHistory.created_at)) - ) - recent_buys = cooldown_check.scalars().all() - if recent_buys and (_get_kst_now() - recent_buys[0].created_at).total_seconds() < cooldown_min * 60: - return {"action": "HOLD", "stock": stock_name, "reason": "cooldown"} - - mcp_ok, mcp_reason = await _mcp_validate_buy(stock) - if not mcp_ok: - return {"action": "HOLD", "stock": stock_name, "reason": mcp_reason} - - current_value = float(holding.quantity * current_price) if holding else 0.0 - regime_str = stock.get("_regime_str", "bear") - regime = MarketRegime.BULL if regime_str == "bull" else MarketRegime.BEAR - invest_amount = calculate_position_size_v2( - total_assets=total_assets, - available_cash=available_cash, - current_strategy_allocation=current_value, - regime=regime, - rsi=stock.get("rsi"), - adx=stock.get("adx_14"), - ) - if invest_amount <= 0 and total_assets <= 0: - invest_amount = available_cash * 0.04 - - quantity = int(invest_amount / current_price) - - if quantity > 0: - try: - if is_domestic: - order_result = await korea_investment_api.place_order(stock_code, "BUY", quantity, int(current_price)) - else: - order_result = await korea_investment_api.place_overseas_order(stock_code, "BUY", quantity, float(current_price), exchange_code) - - total_amount = Decimal(str(current_price)) * quantity - entry_reason = f"매수 신호: {', '.join(signals[:3])}. 점수: {score_val}점" - - success = order_result.get("rt_cd") == "0" - status = "COMPLETED" if success else ("PENDING" if order_result.get("rt_cd") == "1" else "FAILED") - - trade = TradeHistory( - rule_id=stock.get("rule_id"), - stock_code=stock_code, - stock_name=stock_name, - order_type="BUY", - order_price=Decimal(str(current_price)), - order_quantity=quantity, - executed_price=Decimal(str(current_price)) if success else None, - executed_quantity=quantity if success else 0, - total_amount=total_amount, - status=status, - order_id=order_result.get("odno"), - message=f"자동매수: {'추가매수' if holding else '신규'} | {' | '.join(signals[:2])}", - entry_reason=entry_reason, - signals=signals, - created_at=_get_kst_now(), - executed_at=_get_kst_now() if success else None - ) - db.add(trade) - await db.commit() - - if success: - if holding: - new_total_qty = holding.quantity + quantity - prev_cost = float(holding.average_price or 0) * holding.quantity - holding.quantity = new_total_qty - holding.average_price = Decimal(str((prev_cost + current_price * quantity) / new_total_qty)) - holding.current_price = Decimal(str(current_price)) - else: - new_holding = Portfolio( - stock_code=stock_code, stock_name=stock_name, quantity=quantity, - average_price=Decimal(str(current_price)), current_price=Decimal(str(current_price)), - profit_loss=Decimal("0"), profit_loss_rate=Decimal("0") - ) - db.add(new_holding) - await db.commit() - - try: - from app.api.endpoints.portfolio import portfolio_cache - portfolio_cache.apply_buy(stock_code, stock_name, quantity, current_price) - except Exception: - pass - return {"action": "BUY", "stock": stock_name, "quantity": quantity, "price": current_price, "success": success} - - except Exception as e: - return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)} - else: - return {"action": "SKIP", "reason": "quantity_zero", "stock": stock_name} - - elif should_sell and holding: - quantity = holding.quantity - _avg = float(holding.average_price or 0) - profit_loss_rate = ((current_price - _avg) / _avg) * 100 if _avg > 0 else float(holding.profit_loss_rate or 0) - - take_profit = stock.get("take_profit", 3.0) - stop_loss = stock.get("stop_loss", -1.5) - - should_execute_sell = profit_loss_rate >= take_profit or profit_loss_rate <= stop_loss or should_sell - - if should_execute_sell: - try: - if is_domestic: - order_result = await korea_investment_api.place_order(stock_code, "SELL", quantity, int(current_price)) - else: - order_result = await korea_investment_api.place_overseas_order(stock_code, "SELL", quantity, float(current_price), exchange_code) - - recent_cutoff = _get_kst_now() - timedelta(seconds=10) - duplicate_check = await db.execute( - select(TradeHistory).where( - TradeHistory.stock_code == stock_code, - TradeHistory.order_type == "SELL", - TradeHistory.created_at >= recent_cutoff - ) - ) - if duplicate_check.scalar_one_or_none(): - return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name} - - total_amount = Decimal(str(current_price)) * quantity - profit_loss = (current_price - float(holding.average_price or current_price)) * quantity - - success = order_result.get("rt_cd") == "0" - status = "COMPLETED" if success else ("PENDING" if order_result.get("rt_cd") == "1" else "FAILED") - - trade = TradeHistory( - rule_id=stock.get("rule_id"), stock_code=stock_code, stock_name=stock_name, - order_type="SELL", order_price=Decimal(str(current_price)), order_quantity=quantity, - executed_price=Decimal(str(current_price)) if success else None, - executed_quantity=quantity if success else 0, - total_amount=total_amount, status=status, order_id=order_result.get("odno"), - message=f"자동매도: 손익률 {profit_loss_rate:.2f}%", - exit_reason=f"손익: {profit_loss:,.0f}원", - signals=signals, net_profit_loss=Decimal(str(profit_loss)) if success else None, - profit_loss_rate=Decimal(str(profit_loss_rate)) if success else None, - created_at=_get_kst_now(), executed_at=_get_kst_now() if success else None - ) - db.add(trade) - await db.commit() - - if success: - await db.delete(holding) - await db.commit() - try: - from app.api.endpoints.portfolio import portfolio_cache - avg = float(holding.average_price or current_price) - portfolio_cache.apply_sell(stock_code, quantity, current_price, avg) - portfolio_cache.holdings = [h for h in portfolio_cache.holdings if h.stock_code != stock_code] - except Exception: - pass - return {"action": "SELL", "stock": stock_name, "quantity": quantity, "price": current_price, "profit_loss": profit_loss_rate, "success": success} - - except Exception as e: - return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)} - return {"action": "HOLD", "stock": stock_name} except Exception as e: - return {"action": "ERROR", "error": str(e)} \ No newline at end of file + return {"action": "ERROR", "error": str(e)}