#!/usr/bin/env python3
"""
US Stock Tracking and Trading Agent

This module performs buy/sell decisions using AI-based US stock analysis reports
and manages trading records.

Main Features:
1. Generate trading scenarios based on analysis reports
2. Manage stock purchases/sales (maximum 10 slots)
3. Track trading history and returns
4. Share results through Telegram channel

Key Differences from Korean Version:
- Uses ticker symbols (AAPL, MSFT) instead of 6-digit codes
- Uses yfinance for price data
- Uses USD currency
- US market hours (09:30-16:00 EST)
- Uses us_* database tables
"""
from dotenv import load_dotenv
load_dotenv()

import asyncio
import json
import logging
import os
import re
import sqlite3
import sys
import traceback
import importlib.util as _ilu
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Any, Tuple, Optional

# Add parent directory to path for imports
PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
_openai_debug_spec = _ilu.spec_from_file_location("cores.openai_debug", PROJECT_ROOT / "cores" / "openai_debug.py")
if _openai_debug_spec and _openai_debug_spec.loader:
    _openai_debug_mod = _ilu.module_from_spec(_openai_debug_spec)
    _openai_debug_spec.loader.exec_module(_openai_debug_mod)

_error_spec = _ilu.spec_from_file_location("prism_root_openai_error_logging", PROJECT_ROOT / "cores" / "openai_error_logging.py")
if _error_spec and _error_spec.loader:
    _error_mod = _ilu.module_from_spec(_error_spec)
    _error_spec.loader.exec_module(_error_mod)
    log_openai_error = _error_mod.log_openai_error

from telegram import Bot
from telegram.error import TelegramError

# O'Neil 룰베이스 매도 fallback (2026-06-04 quota 사고 대응).
# prism-us/cores 가 sys.path 우선이라 prism-us/cores/oneil_fallback 로 해석됨.
# 방어적 import: 실패 시 _ONEIL_FALLBACK_AVAILABLE=False 로 기존 레거시 룰 유지.
try:
    from cores.oneil_fallback import (
        evaluate_oneil_sell as _oneil_eval,
        from_stock_data as _oneil_from,
    )
    _ONEIL_FALLBACK_AVAILABLE = True
except Exception:  # pragma: no cover - defensive
    _ONEIL_FALLBACK_AVAILABLE = False

# Logging configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler(f"us_stock_tracking_{datetime.now().strftime('%Y%m%d')}.log")
    ]
)
logger = logging.getLogger(__name__)

# MCP related imports
from mcp_agent.app import MCPApp
from mcp_agent.workflows.llm.augmented_llm import RequestParams
import importlib.util as _ilu
_spec = _ilu.spec_from_file_location(
    "openai_responses_llm",
    Path(__file__).resolve().parent.parent / "cores" / "llm" / "openai_responses_llm.py",
)
assert _spec is not None and _spec.loader is not None
_mod = _ilu.module_from_spec(_spec)
_spec.loader.exec_module(_mod)  # type: ignore[union-attr]
OpenAIAugmentedLLM = _mod.OpenAIResponsesLLM
del _ilu, _spec, _mod

# Import US-specific modules
# Use explicit path to avoid conflicts with main project
_prism_us_dir = Path(__file__).parent
sys.path.insert(0, str(_prism_us_dir))


# =============================================================================
# Helper function to import modules from main project cores/ (avoid namespace collision)
# =============================================================================
def _import_from_main_cores(module_name: str, relative_path: str):
    """
    Import module directly from main project cores/ directory.

    This function avoids namespace collision where prism-us/cores/ shadows
    the main project's cores/ directory in sys.path.

    Args:
        module_name: Module name for sys.modules registration
        relative_path: Path relative to PROJECT_ROOT (e.g., "cores/agents/telegram_translator_agent.py")

    Returns:
        Loaded module object
    """
    import importlib.util
    file_path = PROJECT_ROOT / relative_path
    spec = importlib.util.spec_from_file_location(module_name, file_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


# Pre-load telegram_translator_agent from main project (used in multiple methods)
_translator_module = _import_from_main_cores(
    "telegram_translator_agent",
    "cores/agents/telegram_translator_agent.py"
)
translate_telegram_message = _translator_module.translate_telegram_message

# Load parse_llm_json from main project cores/utils.py
# (avoids prism-us/cores/ namespace collision)
_utils_module = _import_from_main_cores("cores_utils", "cores/utils.py")
parse_llm_json = _utils_module.parse_llm_json

try:
    # First try direct import from prism-us directory
    from cores.agents.trading_agents import create_us_trading_scenario_agent, create_us_sell_decision_agent
    from tracking.db_schema import (
        create_us_tables,
        create_us_indexes,
        add_sector_column_if_missing,
        add_market_column_to_shared_tables,
        migrate_us_performance_tracker_columns,
        migrate_us_watchlist_history_columns,
        is_us_ticker_in_holdings,
        get_us_holdings_count,
        get_us_existing_position_for_ticker,
        evaluate_us_pyramid_add_gate,
        compute_us_fractional_sell_quantity,
        decide_us_sell_plan,
    )
    from tracking.journal import USJournalManager
    from tracking.compression import USCompressionManager
except ImportError as e:
    logger.warning(f"Direct import failed: {e}, trying fallback...")
    # Fallback: try adding parent directory
    _prism_us_fallback = Path(__file__).parent
    if str(_prism_us_fallback) not in sys.path:
        sys.path.insert(0, str(_prism_us_fallback))
    from cores.agents.trading_agents import create_us_trading_scenario_agent, create_us_sell_decision_agent
    from tracking.db_schema import (
        create_us_tables,
        create_us_indexes,
        add_sector_column_if_missing,
        add_market_column_to_shared_tables,
        migrate_us_performance_tracker_columns,
        migrate_us_watchlist_history_columns,
        is_us_ticker_in_holdings,
        get_us_holdings_count,
        get_us_existing_position_for_ticker,
        evaluate_us_pyramid_add_gate,
        compute_us_fractional_sell_quantity,
        decide_us_sell_plan,
    )
    from tracking.journal import USJournalManager
    from tracking.compression import USCompressionManager
# Load kis_auth from main project trading/ (prism-us/trading/ has no kis_auth)
import importlib.util as _importlib_util
_kis_auth_spec = _importlib_util.spec_from_file_location("kis_auth", PROJECT_ROOT / "trading/kis_auth.py")
ka = _importlib_util.module_from_spec(_kis_auth_spec)
_kis_auth_spec.loader.exec_module(ka)

# Create MCPApp instance
app = MCPApp(name="us_stock_tracking")


# =============================================================================
# US-Specific Helper Functions
# =============================================================================

def extract_ticker_info(report_path: str) -> Tuple[str, str]:
    """
    Extract ticker and company name from report file path.

    Args:
        report_path: Report file path (e.g., "AAPL_Apple Inc_20260118.pdf")

    Returns:
        Tuple[str, str]: Ticker, company name
    """
    try:
        file_name = Path(report_path).stem
        # Pattern: TICKER_CompanyName_date
        pattern = r'^([A-Z]+)_([^_]+)'
        match = re.match(pattern, file_name)

        if match:
            ticker = match.group(1)
            company_name = match.group(2)
            return ticker, company_name
        else:
            # Fallback
            parts = file_name.split('_')
            if len(parts) >= 2:
                return parts[0], parts[1]

        logger.error(f"Cannot extract ticker info from filename: {file_name}")
        return "", ""
    except Exception as e:
        logger.error(f"Error extracting ticker info: {str(e)}")
        return "", ""


async def get_current_stock_price(cursor, ticker: str, account_key: str | None = None) -> float:
    """
    Get current US stock price using yfinance.

    Args:
        cursor: SQLite cursor
        ticker: Stock ticker symbol (e.g., "AAPL")

    Returns:
        float: Current stock price in USD
    """
    import asyncio
    import yfinance as yf

    # yfinance can intermittently fail/throttle. Retry a few times before falling
    # back, so a momentary blip does not silently drop a fresh buy candidate whose
    # price is not yet in the DB (last-price fallback returns 0 → report skipped).
    # Mirrors the KR tracking/helpers.py retry logic.
    MAX_RETRIES = 3
    for attempt in range(MAX_RETRIES):
        try:
            stock = yf.Ticker(ticker)
            info = stock.info
            current_price = info.get('regularMarketPrice', 0) or info.get('previousClose', 0)

            if current_price > 0:
                logger.info(f"{ticker} current price: ${current_price:.2f}")
                return float(current_price)
            else:
                logger.warning(f"Cannot get price for {ticker}")
                return _get_last_price_from_db(cursor, ticker, account_key=account_key)

        except Exception as e:
            logger.error(f"Error querying current price for {ticker} "
                         f"(attempt {attempt + 1}/{MAX_RETRIES}): {str(e)}")
            if attempt < MAX_RETRIES - 1:
                wait = 2 * (attempt + 1)  # 2s, 4s
                logger.warning(f"{ticker} price query retry in {wait}s")
                await asyncio.sleep(wait)
            else:
                return _get_last_price_from_db(cursor, ticker, account_key=account_key)


def _get_last_price_from_db(cursor, ticker: str, account_key: str | None = None) -> float:
    """Get last saved price from DB as fallback."""
    try:
        if account_key:
            cursor.execute(
                "SELECT current_price FROM us_stock_holdings WHERE ticker = ? AND account_key = ?",
                (ticker, account_key)
            )
        else:
            cursor.execute(
                "SELECT current_price FROM us_stock_holdings WHERE ticker = ?",
                (ticker,)
            )
        row = cursor.fetchone()
        if row and row[0]:
            last_price = float(row[0])
            logger.warning(f"{ticker} price query failed, using last price: ${last_price:.2f}")
            return last_price
    except Exception:
        pass
    return 0.0


async def get_trading_value_rank_change(ticker: str) -> Tuple[float, str]:
    """
    Calculate trading value ranking change for a US stock.

    Args:
        ticker: Stock ticker symbol

    Returns:
        Tuple[float, str]: Ranking change percentage, analysis result message
    """
    try:
        import yfinance as yf

        stock = yf.Ticker(ticker)
        hist = stock.history(period="5d")

        if hist.empty or len(hist) < 2:
            return 0, "Insufficient historical data"

        # Get recent 2 days
        recent_volume = hist['Volume'].iloc[-1]
        previous_volume = hist['Volume'].iloc[-2]
        recent_price = hist['Close'].iloc[-1]
        previous_price = hist['Close'].iloc[-2]

        # Calculate trading value
        recent_value = recent_volume * recent_price
        previous_value = previous_volume * previous_price

        if previous_value > 0:
            value_change_percentage = ((recent_value - previous_value) / previous_value) * 100
        else:
            value_change_percentage = 0

        # Get average volume for context
        avg_volume = hist['Volume'].mean()
        volume_ratio = recent_volume / avg_volume if avg_volume > 0 else 1

        result_msg = (
            f"Trading value: ${recent_value/1e6:.1f}M "
            f"(prev: ${previous_value/1e6:.1f}M, "
            f"change: {'▲' if value_change_percentage > 0 else '▼' if value_change_percentage < 0 else '='}"
            f"{abs(value_change_percentage):.1f}%), "
            f"Volume ratio: {volume_ratio:.2f}x"
        )

        logger.info(f"{ticker} {result_msg}")
        return value_change_percentage, result_msg

    except Exception as e:
        logger.error(f"Error analyzing trading value for {ticker}: {str(e)}")
        return 0, "Trading value analysis failed"


# Apply ratio guard only when portfolio is large enough that the ratio is meaningful.
# With <4 holdings, a single same-sector position naturally produces 25-100% — blocking
# every additional buy in that sector even though absolute count is well under the cap.
MIN_HOLDINGS_FOR_RATIO_CHECK = 4


def check_sector_diversity(cursor, sector: str, max_same_sector: int, concentration_ratio: float, account_key: str | None = None) -> bool:
    """
    Check for over-concentration in same sector.

    The absolute cap (`max_same_sector`) is always enforced. The ratio cap
    (`concentration_ratio`) is only applied once the portfolio holds at least
    `MIN_HOLDINGS_FOR_RATIO_CHECK` positions, so that small portfolios are not
    blocked by trivially high ratios (e.g. 1/2 = 50%).

    Args:
        cursor: SQLite cursor
        sector: GICS sector name
        max_same_sector: Maximum holdings in same sector
        concentration_ratio: Sector concentration limit ratio

    Returns:
        bool: True if can add more, False if over-concentrated
    """
    try:
        if not sector or sector.lower() == "unknown":
            return True

        if account_key:
            cursor.execute("SELECT scenario FROM us_stock_holdings WHERE account_key = ?", (account_key,))
        else:
            cursor.execute("SELECT scenario FROM us_stock_holdings")
        holdings_scenarios = cursor.fetchall()

        sectors = []
        for row in holdings_scenarios:
            if row[0]:
                try:
                    scenario_data = json.loads(row[0])
                    if 'sector' in scenario_data:
                        sectors.append(scenario_data['sector'])
                except Exception:
                    pass

        same_sector_count = sum(1 for s in sectors if s and s.lower() == sector.lower())

        if same_sector_count >= max_same_sector:
            logger.warning(
                f"Sector '{sector}' absolute cap reached: "
                f"holding {same_sector_count} stocks (max {max_same_sector})"
            )
            return False

        if len(sectors) >= MIN_HOLDINGS_FOR_RATIO_CHECK and \
           same_sector_count / len(sectors) >= concentration_ratio:
            logger.warning(
                f"Sector '{sector}' ratio cap reached: "
                f"{same_sector_count}/{len(sectors)} = "
                f"{same_sector_count/len(sectors)*100:.0f}% "
                f"(limit {concentration_ratio*100:.0f}%)"
            )
            return False

        return True

    except Exception as e:
        logger.error(f"Error checking sector diversity: {str(e)}")
        return True


def parse_price_value(value: Any) -> float:
    """Parse price value and convert to number."""
    try:
        if isinstance(value, (int, float)):
            return float(value)

        if isinstance(value, str):
            value = value.replace(',', '').replace('$', '')

            range_patterns = [
                r'(\d+(?:\.\d+)?)\s*[-~]\s*(\d+(?:\.\d+)?)',
            ]

            for pattern in range_patterns:
                match = re.search(pattern, value)
                if match:
                    low = float(match.group(1))
                    high = float(match.group(2))
                    return (low + high) / 2

            number_match = re.search(r'(\d+(?:\.\d+)?)', value)
            if number_match:
                return float(number_match.group(1))

        return 0
    except Exception as e:
        logger.warning(f"Failed to parse price value: {value} - {str(e)}")
        return 0


def default_scenario() -> Dict[str, Any]:
    """Return default trading scenario for US stocks.

    NOTE: This is a *failure* sentinel, not a real trading decision. The
    `analysis_failed` flag lets downstream code distinguish a genuine
    "no_entry" judgment from an analysis/LLM failure, so we never broadcast a
    misleading "매수 보류" message for a stock we couldn't actually analyze.
    """
    return {
        "portfolio_analysis": "Analysis failed",
        "buy_score": 0,
        "decision": "no_entry",
        "target_price": 0,
        "stop_loss": 0,
        "investment_period": "short",
        "rationale": "Analysis failed",
        "sector": "Unknown",
        "considerations": "Analysis failed",
        "analysis_failed": True
    }


# =============================================================================
# US Stock Tracking Agent
# =============================================================================

class USStockTrackingAgent:
    """US Stock Tracking and Trading Agent"""

    # Constants
    MAX_SLOTS = 10  # Maximum number of stocks to hold
    MAX_SAME_SECTOR = 3  # Maximum holdings in same sector
    SECTOR_CONCENTRATION_RATIO = 0.3  # Sector concentration limit ratio

    # Investment period constants
    PERIOD_SHORT = "short"  # Within 1 month
    PERIOD_MEDIUM = "medium"  # 1-3 months
    PERIOD_LONG = "long"  # 3+ months

    # Buy score thresholds
    SCORE_STRONG_BUY = 8  # Strong buy
    SCORE_CONSIDER = 7  # Consider buying
    SCORE_UNSUITABLE = 6  # Unsuitable for buying

    def __init__(
        self,
        db_path: str = "stock_tracking_db.sqlite",
        telegram_token: str = None,
        enable_journal: bool = None
    ):
        """
        Initialize US Stock Tracking Agent.

        Args:
            db_path: SQLite database file path
            telegram_token: Telegram bot token
            enable_journal: Enable trading journal feature.
                Priority: parameter > ENABLE_TRADING_JOURNAL env > default(False).
                (KR 에이전트와 동일 토글 — 기존엔 US가 env 무시하고 False 하드코딩이라 일지 미동작)
        """
        self.max_slots = self.MAX_SLOTS
        self.message_queue = []
        self._msg_types = []  # msg_type for each message in queue
        self._broadcast_task = None  # Track broadcast translation task
        self.trading_agent = None
        self.sell_decision_agent = None
        self.db_path = db_path
        self.conn = None
        self.cursor = None
        self.language = "en"  # Default to English for US
        # Trading journal feature flag — Priority: parameter > env > default(False).
        # KR 에이전트와 동일하게 ENABLE_TRADING_JOURNAL env 를 존중한다.
        if enable_journal is not None:
            self.enable_journal = enable_journal
        else:
            env_value = os.environ.get("ENABLE_TRADING_JOURNAL", "false").lower()
            self.enable_journal = env_value in ("true", "1", "yes")
        self.account_configs: list[dict[str, Any]] = []
        self.active_account: dict[str, Any] | None = None

        # Journal and compression managers (initialized in initialize())
        self.journal_manager = None
        self.compression_manager = None

        # Set Telegram bot token
        self.telegram_token = telegram_token or os.environ.get("TELEGRAM_BOT_TOKEN")
        self.telegram_bot = None
        if self.telegram_token:
            self.telegram_bot = Bot(token=self.telegram_token)

    async def initialize(self, language: str = "ko", sector_names: list = None,
                         skip_llm_agent: bool = False):
        """
        Create necessary tables and initialize.

        Args:
            language: Language code for agents (default: "ko")
            sector_names: List of valid sector names for trading agent (optional)
            skip_llm_agent: When True, skip creating the LLM trading-scenario agent.
                The sell path (sell_stock / send_telegram_message) does NOT use
                self.trading_agent, so the LLM-free Loop A hard-stop loop can reuse
                the sell/journal/telegram plumbing without the heavy LLM agent.
                Default False keeps batch behaviour byte-for-byte unchanged.
        """
        logger.info("Starting US tracking agent initialization")

        self.language = language

        # Initialize SQLite connection
        self.conn = sqlite3.connect(self.db_path)
        self.conn.row_factory = sqlite3.Row
        self.cursor = self.conn.cursor()

        # Initialize trading scenario agent for US (skipped for lightweight consumers).
        self.trading_agent = None if skip_llm_agent else \
            create_us_trading_scenario_agent(language=language, sector_names=sector_names)

        # Initialize sell decision agent for US
        self.sell_decision_agent = create_us_sell_decision_agent(language=language)

        # Create US database tables
        await self._create_tables()

        # Initialize journal manager
        self.journal_manager = USJournalManager(
            cursor=self.cursor,
            conn=self.conn,
            language=language,
            enable_journal=self.enable_journal
        )

        # Initialize compression manager
        self.compression_manager = USCompressionManager(
            cursor=self.cursor,
            conn=self.conn
        )
        self.account_configs = self._get_trading_accounts()
        if self.account_configs:
            self._set_active_account(self.account_configs[0])
        else:
            logger.warning("No trading accounts configured - skipping trade execution")

        logger.info(f"US tracking agent initialization complete (journal: {self.enable_journal})")
        return True

    async def _create_tables(self):
        """Create necessary US database tables."""
        create_us_tables(self.cursor, self.conn)
        create_us_indexes(self.cursor, self.conn)
        add_sector_column_if_missing(self.cursor, self.conn)
        # Add market column to shared tables for KR/US distinction
        add_market_column_to_shared_tables(self.cursor, self.conn)
        # Migrate performance tracker columns (tracking_status, was_traded, etc.)
        migrate_us_performance_tracker_columns(self.cursor, self.conn)
        # Migrate watchlist history columns for 7/14/30-day performance tracking
        migrate_us_watchlist_history_columns(self.cursor, self.conn)

    def _get_trading_accounts(self) -> List[Dict[str, Any]]:
        default_mode = str(ka.getEnv().get("default_mode", "demo")).strip().lower()
        svr = "vps" if default_mode == "demo" else "prod"
        return ka.get_configured_accounts(svr=svr, market="us")

    def _set_active_account(self, account: Dict[str, Any]) -> None:
        self.active_account = account

    def _require_active_account(self) -> Dict[str, Any]:
        if not self.active_account:
            raise RuntimeError("No active US trading account is set")
        return self.active_account

    def _account_scope(self) -> Tuple[str, str]:
        account = self._require_active_account()
        return account["account_key"], account["name"]

    @staticmethod
    def _safe_account_log_label(account: Dict[str, Any]) -> str:
        """Format account identity for logs without exposing raw account numbers."""
        account_name = account.get("name", "unknown")
        account_key = str(account.get("account_key", "") or "")
        if not account_key:
            return account_name

        parts = account_key.split(":")
        if len(parts) == 3:
            scope, account_number, product = parts
            return f"{account_name} ({scope}:{ka.mask_account_number(account_number)}:{product})"

        return f"{account_name} ({ka.mask_account_number(account_key)})"

    def _normalize_decision(self, decision: str) -> str:
        """
        Normalize decision string for comparison.

        The agent prompt uses "Enter" or "No Entry" but code checks may use
        lowercase variants. This method normalizes all variants to a consistent format.

        Args:
            decision: Raw decision string from agent

        Returns:
            str: Normalized decision ("entry" or "no_entry")
        """
        if not decision:
            return "no_entry"
        d = decision.lower().strip()
        # Handle various entry formats
        if d in ("enter", "entry", "진입", "yes", "buy"):
            return "entry"
        # Handle various no-entry formats
        elif d in ("no entry", "no_entry", "no-entry", "미진입", "no", "skip", "pass"):
            return "no_entry"
        return d

    async def _extract_ticker_info(self, report_path: str) -> Tuple[str, str]:
        """Extract ticker and company name from report path."""
        return extract_ticker_info(report_path)

    async def _get_current_stock_price(self, ticker: str) -> float:
        """Get current stock price."""
        account_key, _ = self._account_scope()
        return await get_current_stock_price(self.cursor, ticker, account_key=account_key)

    async def _get_trading_value_rank_change(self, ticker: str) -> Tuple[float, str]:
        """Calculate trading value ranking change."""
        return await get_trading_value_rank_change(ticker)

    async def _is_ticker_in_holdings(self, ticker: str) -> bool:
        """Check if stock is already in holdings."""
        account_key, _ = self._account_scope()
        return is_us_ticker_in_holdings(self.cursor, ticker, account_key=account_key)

    async def _get_current_slots_count(self) -> int:
        """Get current number of holdings."""
        account_key, _ = self._account_scope()
        return get_us_holdings_count(self.cursor, account_key=account_key)

    async def _check_sector_diversity(self, sector: str, is_pyramiding_add: bool = False) -> bool:
        """Check for over-concentration in same sector.

        Pyramiding adds (#288) target a ticker already held, so the name is
        already counted toward the sector. Adding to it does not introduce a
        new name into the sector, so the count-based concentration limit should
        not block it — only brand-new positions are subject to the limit.
        """
        if is_pyramiding_add:
            return True
        account_key, _ = self._account_scope()
        return check_sector_diversity(
            self.cursor, sector,
            self.MAX_SAME_SECTOR, self.SECTOR_CONCENTRATION_RATIO, account_key=account_key
        )

    async def _extract_trading_scenario(
        self,
        report_content: str,
        rank_change_msg: str = "",
        ticker: str = None,
        sector: str = None,
        trigger_type: str = "",
        trigger_mode: str = ""
    ) -> Dict[str, Any]:
        """
        Extract trading scenario from report.

        Args:
            report_content: Analysis report content
            rank_change_msg: Trading value ranking change info
            ticker: Stock ticker symbol
            sector: Stock sector
            trigger_type: Trigger type
            trigger_mode: Trigger mode

        Returns:
            Dict: Trading scenario information
        """
        try:
            # Get current holdings info
            current_slots = await self._get_current_slots_count()

            # Collect current portfolio information
            self.cursor.execute("""
                SELECT ticker, company_name, buy_price, current_price, scenario
                FROM us_stock_holdings
                WHERE account_key = ?
            """, (self._account_scope()[0],))
            holdings = [dict(row) for row in self.cursor.fetchall()]

            # Analyze sector distribution
            sector_distribution = {}
            investment_periods = {"short": 0, "medium": 0, "long": 0}

            for holding in holdings:
                scenario_str = holding.get('scenario', '{}')
                try:
                    if isinstance(scenario_str, str):
                        scenario_data = json.loads(scenario_str)
                        sector_name = scenario_data.get('sector', 'Unknown')
                        sector_distribution[sector_name] = sector_distribution.get(sector_name, 0) + 1
                        period = scenario_data.get('investment_period', 'medium')
                        investment_periods[period] = investment_periods.get(period, 0) + 1
                except Exception:
                    pass

            # Portfolio info string
            portfolio_info = f"""
            Current holdings: {current_slots}/{self.max_slots}
            Sector distribution: {json.dumps(sector_distribution, ensure_ascii=False)}
            Investment period distribution: {json.dumps(investment_periods, ensure_ascii=False)}
            """

            # Get trading journal context for informed decisions
            journal_context = ""
            score_adjustment_info = ""
            adjustment, reasons = 0, []
            if ticker:
                journal_context = self.get_journal_context(
                    ticker=ticker,
                    sector=sector,
                    trigger_type=trigger_type
                )
                if journal_context:
                    logger.info(f"[Journal] US injected context for {ticker} ({len(journal_context)} chars)")
                    logger.debug(f"[Journal] US context preview: {journal_context[:500]}")
                elif self.enable_journal:
                    logger.warning(f"[Journal] US empty context for {ticker} despite journal being enabled")
                else:
                    logger.debug(f"[Journal] US journal disabled, no context for {ticker}")
                # Get score adjustment suggestion
                adjustment, reasons = self.get_score_adjustment(ticker, sector, trigger_type)
                if adjustment != 0 or reasons:
                    score_adjustment_info = f"""
                ### 📊 Score Adjustment Suggestion (Experience-Based)
                - Recommended Adjustment: {'+' if adjustment > 0 else ''}{adjustment} points
                - Reason: {', '.join(reasons) if reasons else 'N/A'}
                - ⚠️ This adjustment is a reference based on past experience.
                """

            # LLM call to generate trading scenario
            llm = await self.trading_agent.attach_llm(OpenAIAugmentedLLM)

            # Build trigger info section
            trigger_info_section = ""
            if trigger_type:
                trigger_info_section = f"""
                ### Trigger Info (Apply Trigger-Based Entry Criteria)
                - **Triggered By**: {trigger_type}
                - **Trigger Mode**: {trigger_mode or 'unknown'}
                """

            prompt_message = f"""
            This is an AI analysis report for a US stock. Please generate a trading scenario based on this report.

            ### Current Portfolio Status:
            {portfolio_info}
            {trigger_info_section}
            ### Trading Value Analysis:
            {rank_change_msg}
            {score_adjustment_info}
            {journal_context}

            ### Report Content:
            {report_content}
            """

            # LLM scenario generation with retry+backoff. Transient API/parse
            # failures (rate limits, truncated responses) were silently turning
            # into default_scenario() — i.e. a misleading "Analysis failed" skip.
            # Retry a couple of times before giving up.
            ticker_tag = ticker or "?"
            max_attempts = 3
            response = None
            scenario_json = None
            for attempt in range(1, max_attempts + 1):
                try:
                    response = await llm.generate_str(
                        message=prompt_message,
                        request_params=RequestParams(
                            model="gpt-5.5",
                            maxTokens=30000
                        )
                    )
                    # JSON parsing (consolidated in cores/utils.py)
                    scenario_json = parse_llm_json(response, context='US trading scenario')
                    if scenario_json is not None:
                        break
                    logger.warning(
                        f"[{ticker_tag}] US trading scenario parse returned None "
                        f"(attempt {attempt}/{max_attempts})"
                    )
                except Exception as call_err:
                    log_openai_error(logger, call_err, f"US trading scenario LLM call [{ticker_tag}]")
                    logger.warning(
                        f"[{ticker_tag}] US trading scenario LLM call failed "
                        f"(attempt {attempt}/{max_attempts}): {call_err}"
                    )
                if attempt < max_attempts:
                    await asyncio.sleep(2 * attempt)  # linear backoff: 2s, 4s

            if scenario_json is not None:
                # Persist the experience-based score adjustment alongside the scenario.
                # It rides inside the scenario JSON, stored in us_stock_holdings.scenario and
                # copied to us_trading_history.scenario on sell — giving the weekly influence
                # report a journal-impact signal for free (#280).
                if adjustment != 0 or reasons:
                    scenario_json["score_adjustment"] = {"value": adjustment, "reasons": reasons}
                logger.info(f"Scenario parsed: {json.dumps(scenario_json, ensure_ascii=False)[:200]}")
                return scenario_json

            logger.error(
                f"[ANALYSIS_FAILED][{ticker_tag}] US trading scenario unavailable after "
                f"{max_attempts} attempts. Last response (truncated): {str(response)[:500]}"
            )
            return default_scenario()

        except Exception as e:
            log_openai_error(logger, e, "US trading scenario extraction")
            logger.error(f"Error extracting trading scenario: {str(e)}")
            logger.error(traceback.format_exc())
            return default_scenario()

    async def _analyze_report_core(self, pdf_report_path: str) -> Dict[str, Any]:
        """Analyze a report once before per-account execution checks.

        Note:
            `_extract_trading_scenario()` includes the currently active account's
            portfolio state in the LLM context. In multi-account mode this means
            the primary account shapes the shared report analysis, while actual
            buy eligibility is still re-checked per account in `process_reports()`.
            This keeps LLM cost flat instead of multiplying per account.
        """
        try:
            logger.info(f"Starting report analysis: {pdf_report_path}")

            ticker, company_name = await self._extract_ticker_info(pdf_report_path)
            if not ticker or not company_name:
                logger.error(f"Failed to extract ticker info: {pdf_report_path}")
                return {"success": False, "error": "Failed to extract ticker info"}

            current_price = await self._get_current_stock_price(ticker)
            if current_price <= 0:
                logger.error(f"{ticker} current price query failed")
                return {"success": False, "error": "Current price query failed"}

            rank_change_percentage, rank_change_msg = await self._get_trading_value_rank_change(ticker)

            from pdf_converter import pdf_to_markdown_text

            report_content = pdf_to_markdown_text(pdf_report_path)
            trigger_info = getattr(self, 'trigger_info_map', {}).get(ticker, {})
            trigger_type = trigger_info.get('trigger_type', '')
            trigger_mode = trigger_info.get('trigger_mode', '')

            scenario = await self._extract_trading_scenario(
                report_content,
                rank_change_msg,
                ticker=ticker,
                sector=None,
                trigger_type=trigger_type,
                trigger_mode=trigger_mode
            )

            # Analysis/LLM failure sentinel: do NOT emit a misleading "매수 보류"
            # message or watchlist entry. Log clearly and skip the ticker via the
            # same path as a hard failure (success=False), unifying both failure modes.
            if scenario.get("analysis_failed"):
                logger.error(
                    f"[ANALYSIS_FAILED][{ticker}] {company_name}: trading scenario unavailable "
                    f"after retries — skipping (no trade, no skip message, no watchlist)"
                )
                return {"success": False, "error": "analysis_failed",
                        "ticker": ticker, "company_name": company_name}

            raw_decision = scenario.get("decision", "no_entry")
            sector = scenario.get("sector", "Unknown")

            return {
                "success": True,
                "ticker": ticker,
                "company_name": company_name,
                "current_price": current_price,
                "scenario": scenario,
                "decision": self._normalize_decision(raw_decision),
                "raw_decision": raw_decision,
                "sector": sector,
                "rank_change_percentage": rank_change_percentage,
                "rank_change_msg": rank_change_msg
            }

        except Exception as e:
            logger.error(f"[ANALYSIS_FAILED] Error analyzing report ({pdf_report_path}): {str(e)}")
            logger.error(traceback.format_exc())
            return {"success": False, "error": str(e)}

    async def analyze_report(self, pdf_report_path: str) -> Dict[str, Any]:
        """
        Analyze US stock analysis report and make trading decision.

        Args:
            pdf_report_path: PDF analysis report file path

        Returns:
            Dict: Trading decision result
        """
        analysis_result = await self._analyze_report_core(pdf_report_path)
        if not analysis_result.get("success", False):
            return analysis_result

        ticker = analysis_result.get("ticker")
        company_name = analysis_result.get("company_name")
        if await self._is_ticker_in_holdings(ticker):
            # Pyramiding (#288): allow an additional independent entry only when the
            # strong-bull add-gate passes. Otherwise keep the legacy hard block.
            scenario = analysis_result.get("scenario", {}) or {}
            current_price = analysis_result.get("current_price", 0)
            account_key = self._account_scope()[0]
            existing = get_us_existing_position_for_ticker(self.cursor, ticker, account_key=account_key)
            allowed, reason = evaluate_us_pyramid_add_gate(
                market_condition=scenario.get("market_condition", ""),
                existing_avg_buy_price=existing.get("avg_buy_price", 0.0),
                current_price=current_price,
                existing_row_count=existing.get("row_count", 0),
            )
            if not allowed:
                logger.info(f"{ticker} ({company_name}) already in holdings — add gate blocked: {reason}")
                return {
                    "success": True,
                    "decision": "holding",
                    "ticker": ticker,
                    "company_name": company_name,
                    "current_price": current_price
                }
            logger.info(f"{ticker} ({company_name}) pyramiding add gate passed: {reason}")
            analysis_result["is_add"] = True
            analysis_result["existing_avg_buy_price"] = existing.get("avg_buy_price", 0.0)
            analysis_result["existing_row_count"] = existing.get("row_count", 0)

        analysis_result["sector_diverse"] = await self._check_sector_diversity(
            analysis_result.get("sector", "Unknown")
        )
        return analysis_result

    async def buy_stock(self, ticker: str, company_name: str, current_price: float,
                        scenario: Dict[str, Any], rank_change_msg: str = "", is_add: bool = False) -> bool:
        """
        Process stock purchase.

        Args:
            ticker: Stock ticker symbol
            company_name: Company name
            current_price: Current stock price in USD
            scenario: Trading scenario information
            rank_change_msg: Trading value ranking change info
            is_add: Pyramiding add (#288) — bypass the already-holding re-check and
                    insert an independent additional row instead of a first entry.

        Returns:
            bool: Purchase success status
        """
        try:
            # Check if already holding (skipped for a pyramiding add)
            if not is_add and await self._is_ticker_in_holdings(ticker):
                logger.warning(f"{ticker} ({company_name}) already in holdings")
                return False

            # Check available slots
            current_slots = await self._get_current_slots_count()
            if current_slots >= self.max_slots:
                logger.warning(f"Holdings already at maximum ({self.max_slots})")
                return False

            # Current time
            now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            account_key, account_name = self._account_scope()

            # Get trigger info
            trigger_info = getattr(self, 'trigger_info_map', {}).get(ticker, {})
            trigger_type = trigger_info.get('trigger_type', 'AI_Analysis')
            trigger_mode = trigger_info.get('trigger_mode', getattr(self, 'trigger_mode', 'unknown'))

            # Add to holdings table
            self.cursor.execute(
                """
                INSERT INTO us_stock_holdings
                (account_key, account_name, ticker, company_name, buy_price, buy_date, current_price, last_updated,
                 scenario, target_price, stop_loss, trigger_type, trigger_mode, sector)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    account_key,
                    account_name,
                    ticker,
                    company_name,
                    current_price,
                    now,
                    current_price,
                    now,
                    json.dumps(scenario, ensure_ascii=False),
                    scenario.get('target_price', 0),
                    scenario.get('stop_loss', 0),
                    trigger_type,
                    trigger_mode,
                    scenario.get('sector', 'Unknown')
                )
            )
            self.conn.commit()

            # Build buy message (same format as KR template).
            # Pyramiding adds (#288) get a distinct header showing the entry number
            # and the new aggregate average price.
            target_price = scenario.get('target_price', 0)
            stop_loss = scenario.get('stop_loss', 0)

            if is_add:
                agg = get_us_existing_position_for_ticker(self.cursor, ticker, account_key=account_key)
                entry_no = agg.get("row_count", 1)  # this entry is the Nth row
                new_avg = agg.get("avg_buy_price", current_price)
                message = f"📈 Add-On Entry (#{entry_no}): {company_name}({ticker})\n" \
                          f"This Entry: ${current_price:,.2f}\n" \
                          f"New Avg Price: ${new_avg:,.2f}\n" \
                          f"⚠️ Portfolio weight increased (1 independent slot consumed)\n" \
                          f"Target: ${target_price:,.2f}\n" \
                          f"Stop Loss: ${stop_loss:,.2f}\n" \
                          f"Period: {scenario.get('investment_period', 'short')}\n" \
                          f"Sector: {scenario.get('sector', 'Unknown')}\n"
            else:
                message = f"📈 New Buy: {company_name}({ticker})\n" \
                          f"Buy Price: ${current_price:,.2f}\n" \
                          f"Target: ${target_price:,.2f}\n" \
                          f"Stop Loss: ${stop_loss:,.2f}\n" \
                          f"Period: {scenario.get('investment_period', 'short')}\n" \
                          f"Sector: {scenario.get('sector', 'Unknown')}\n"

            # Add trigger win rate
            trigger_win_rate = self._get_trigger_win_rate(trigger_type)
            if trigger_win_rate:
                message += f"{trigger_win_rate}\n"

            # Add valuation analysis
            if scenario.get('valuation_analysis'):
                message += f"Valuation: {scenario.get('valuation_analysis')}\n"

            # Add sector outlook (same as KR version)
            if scenario.get('sector_outlook'):
                message += f"Sector Outlook: {scenario.get('sector_outlook')}\n"

            # Add trading value analysis
            if rank_change_msg:
                message += f"Trading Value Analysis: {rank_change_msg}\n"

            message += f"Rationale: {scenario.get('rationale', 'No information')}\n"

            # Surface journal-grounded reasoning so the feedback loop is transparent (#280).
            # All fields optional — defends against scenarios without journal_reflection.
            _jr = scenario.get('journal_reflection') or {}
            if isinstance(_jr, dict):
                if _jr.get('recent_exit_caution'):
                    message += f"⚠️ 최근 매도 주의: {_jr.get('recent_exit_caution')}\n"
                if _jr.get('applied_lessons'):
                    message += f"📒 매매일지 반영: {_jr.get('applied_lessons')}\n"
            _sadj = scenario.get('score_adjustment') or {}
            if isinstance(_sadj, dict) and _sadj.get('value'):
                _rsn = ', '.join(_sadj.get('reasons', []) or [])
                message += f"📊 경험 기반 점수조정: {_sadj.get('value'):+d}점 ({_rsn})\n"

            # Trading scenario details (same format as KR version)
            trading_scenarios = scenario.get('trading_scenarios', {})
            if trading_scenarios and isinstance(trading_scenarios, dict):
                message += "\n" + "="*40 + "\n"
                message += "📋 Trading Scenario\n"
                message += "="*40 + "\n\n"

                # 1. Key Price Levels
                key_levels = trading_scenarios.get('key_levels', {})
                if key_levels:
                    message += "💰 Key Price Levels:\n"

                    # Resistance levels
                    primary_resistance = parse_price_value(key_levels.get('primary_resistance', 0))
                    secondary_resistance = parse_price_value(key_levels.get('secondary_resistance', 0))
                    if primary_resistance or secondary_resistance:
                        message += "  📈 Resistance:\n"
                        if secondary_resistance:
                            message += f"    • 2차: ${secondary_resistance:,.2f}\n"
                        if primary_resistance:
                            message += f"    • 1차: ${primary_resistance:,.2f}\n"

                    # Current price display
                    message += f"  ━━ 현재가: ${current_price:,.2f} ━━\n"

                    # Support levels
                    primary_support = parse_price_value(key_levels.get('primary_support', 0))
                    secondary_support = parse_price_value(key_levels.get('secondary_support', 0))
                    if primary_support or secondary_support:
                        message += "  📉 Support:\n"
                        if primary_support:
                            message += f"    • 1차: ${primary_support:,.2f}\n"
                        if secondary_support:
                            message += f"    • 2차: ${secondary_support:,.2f}\n"

                    # Volume baseline
                    volume_baseline = key_levels.get('volume_baseline', '')
                    if volume_baseline:
                        message += f"  📊 Volume Baseline: {volume_baseline}\n"

                    message += "\n"

                # 2. Sell Signals
                sell_triggers = trading_scenarios.get('sell_triggers', [])
                if sell_triggers:
                    message += "🔔 Sell Signals:\n"
                    for i, trigger in enumerate(sell_triggers, 1):
                        # Select emoji based on condition type
                        if any(kw in trigger.lower() for kw in ["익절", "목표", "저항", "profit", "target", "resistance"]):
                            emoji = "✅"
                        elif any(kw in trigger.lower() for kw in ["stop", "support", "down"]):
                            emoji = "⛔"
                        elif any(kw in trigger.lower() for kw in ["시간", "횡보", "time", "sideways"]):
                            emoji = "⏰"
                        else:
                            emoji = "•"

                        message += f"  {emoji} {trigger}\n"
                    message += "\n"

                # 3. Hold Conditions
                hold_conditions = trading_scenarios.get('hold_conditions', [])
                if hold_conditions:
                    message += "✋ 보유 지속 조건:\n"
                    for condition in hold_conditions:
                        message += f"  • {condition}\n"
                    message += "\n"

                # 4. Portfolio Context
                portfolio_context = trading_scenarios.get('portfolio_context', '')
                if portfolio_context:
                    message += f"💼 포트폴리오 관점:\n  {portfolio_context}\n"

            self._msg_types.append("analysis")
            self.message_queue.append(message)
            logger.info(f"{ticker} ({company_name}) purchase complete")

            return True

        except Exception as e:
            logger.error(f"{ticker} Error during purchase: {str(e)}")
            logger.error(traceback.format_exc())
            return False

    async def _save_watchlist_item(
        self,
        ticker: str,
        company_name: str,
        current_price: float,
        buy_score: int,
        min_score: int,
        decision: str,
        skip_reason: str,
        scenario: Dict[str, Any],
        sector: str,
        was_traded: bool = False
    ) -> bool:
        """
        Save stocks not purchased to us_watchlist_history table and us_analysis_performance_tracker.

        This enables 7/14/30-day performance tracking for analyzed but not entered stocks.

        Args:
            ticker: Stock ticker symbol (e.g., "AAPL")
            company_name: Company name
            current_price: Current price in USD
            buy_score: Buy score from agent
            min_score: Minimum required score
            decision: Decision (entry/no_entry)
            skip_reason: Reason for not entering
            scenario: Complete scenario information
            sector: GICS sector
            was_traded: Whether the stock was actually traded

        Returns:
            bool: Save success status
        """
        try:
            # Current time
            now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

            # Extract necessary information from scenario
            target_price = scenario.get('target_price', 0)
            stop_loss = scenario.get('stop_loss', 0)
            investment_period = scenario.get('investment_period', 'short')
            portfolio_analysis = scenario.get('portfolio_analysis', '')
            valuation_analysis = scenario.get('valuation_analysis', '')
            sector_outlook = scenario.get('sector_outlook', '')
            market_condition = scenario.get('market_condition', '')
            rationale = scenario.get('rationale', '')

            # Get trigger info from parent's trigger_info_map
            trigger_info = getattr(self, 'trigger_info_map', {}).get(ticker, {})
            trigger_type = trigger_info.get('trigger_type', '')
            trigger_mode = trigger_info.get('trigger_mode', '')
            risk_reward_ratio = trigger_info.get('risk_reward_ratio', scenario.get('risk_reward_ratio', 0))

            # Save to us_watchlist_history with trigger info
            self.cursor.execute(
                """
                INSERT INTO us_watchlist_history
                (ticker, company_name, current_price, analyzed_date, buy_score, min_score,
                 decision, skip_reason, target_price, stop_loss, investment_period, sector,
                 scenario, portfolio_analysis, valuation_analysis, sector_outlook,
                 market_condition, rationale, trigger_type, trigger_mode, risk_reward_ratio, was_traded)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    ticker,
                    company_name,
                    current_price,
                    now,
                    buy_score,
                    min_score,
                    decision,
                    skip_reason,
                    target_price,
                    stop_loss,
                    investment_period,
                    sector,
                    json.dumps(scenario, ensure_ascii=False),
                    portfolio_analysis,
                    valuation_analysis,
                    sector_outlook,
                    market_condition,
                    rationale,
                    trigger_type,
                    trigger_mode,
                    risk_reward_ratio,
                    1 if was_traded else 0
                )
            )

            # Also save to us_analysis_performance_tracker for 7/14/30-day tracking
            # Note: US version doesn't use watchlist_id FK (independent design)
            self.cursor.execute(
                """
                INSERT INTO us_analysis_performance_tracker
                (ticker, company_name, analysis_date, analysis_price,
                 predicted_direction, target_price, stop_loss, buy_score,
                 decision, skip_reason, risk_reward_ratio,
                 trigger_type, trigger_mode, sector,
                 tracking_status, was_traded, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)
                """,
                (
                    ticker,
                    company_name,
                    now,
                    current_price,
                    'UP' if target_price > current_price else 'DOWN' if target_price < current_price else 'NEUTRAL',
                    target_price,
                    stop_loss,
                    buy_score,
                    decision,
                    skip_reason,
                    risk_reward_ratio,
                    trigger_type,
                    trigger_mode,
                    sector,
                    1 if was_traded else 0,
                    now
                )
            )

            self.conn.commit()

            # Translate market regime labels to Korean for display
            _regime_labels_ko = {
                "parabolic": "폭주 강세장",
                "strong_bull": "강한 강세장", "moderate_bull": "보통 강세장",
                "sideways": "횡보장", "moderate_bear": "보통 약세장", "strong_bear": "강한 약세장"
            }
            market_condition_display = market_condition
            for eng, ko in _regime_labels_ko.items():
                if market_condition_display.startswith(eng):
                    market_condition_display = market_condition_display.replace(eng, ko, 1)
                    break

            # Generate no-entry message (same format as Korean enhanced version)
            skip_message = f"⚠️ 매수 보류: {company_name}({ticker})\n" \
                           f"현재가: ${current_price:,.2f}\n" \
                           f"매수 Score: {buy_score}/10\n" \
                           f"결정: Skip\n" \
                           f"시장 상황: {market_condition_display}\n" \
                           f"산업군: {sector}\n" \
                           f"보류 사유: {skip_reason}\n" \
                           f"분석 의견: {rationale if rationale else '정보 없음'}"

            # Add trigger win rate
            trigger_win_rate = self._get_trigger_win_rate(trigger_type)
            if trigger_win_rate:
                skip_message += f"\n{trigger_win_rate}"

            # Surface journal-grounded reasoning so the feedback loop is transparent (#280).
            # All fields optional — defends against scenarios without journal_reflection.
            _jr = scenario.get('journal_reflection') or {}
            if isinstance(_jr, dict):
                if _jr.get('recent_exit_caution'):
                    skip_message += f"\n⚠️ 최근 매도 주의: {_jr.get('recent_exit_caution')}"
                if _jr.get('applied_lessons'):
                    skip_message += f"\n📒 매매일지 반영: {_jr.get('applied_lessons')}"
            _sadj = scenario.get('score_adjustment') or {}
            if isinstance(_sadj, dict) and _sadj.get('value'):
                _rsn = ', '.join(_sadj.get('reasons', []) or [])
                skip_message += f"\n📊 경험 기반 점수조정: {_sadj.get('value'):+d}점 ({_rsn})"

            self._msg_types.append("analysis")
            self.message_queue.append(skip_message)

            logger.info(
                f"{ticker}({company_name}) Watchlist save complete - "
                f"Score: {buy_score}/{min_score}, Reason: {skip_reason}, Trigger: {trigger_type}"
            )
            return True

        except Exception as e:
            logger.error(f"{ticker} Error saving watchlist: {str(e)}")
            logger.error(traceback.format_exc())
            return False

    async def _analyze_sell_decision(self, stock_data: Dict[str, Any]) -> Tuple[bool, str]:
        """AI agent-based sell decision analysis.

        Calls sell_decision_agent (LLM) to comprehensively analyze technical trend,
        market conditions, and portfolio balance. Falls back to rule-based logic on error.

        Args:
            stock_data: Stock information

        Returns:
            Tuple[bool, str]: Whether to sell, sell reason
        """
        ticker = stock_data.get('ticker', '')
        company_name = stock_data.get('company_name', '')
        buy_price = stock_data.get('buy_price', 0)
        buy_date = stock_data.get('buy_date', '')
        current_price = stock_data.get('current_price', 0)
        target_price = stock_data.get('target_price', 0)
        stop_loss = stock_data.get('stop_loss', 0)

        try:
            profit_rate = ((current_price - buy_price) / buy_price) * 100 if buy_price > 0 else 0
            buy_datetime = datetime.strptime(buy_date, "%Y-%m-%d %H:%M:%S")
            days_passed = (datetime.now() - buy_datetime).days

            scenario_str = stock_data.get('scenario', '{}')
            period = "medium"
            sector = "Unknown"
            trading_scenarios = {}
            highest_price = max(buy_price, current_price)  # Default to max of buy/current
            highest_price_initialized = False
            initial_stop_loss = stop_loss
            initial_target_price = target_price
            try:
                if isinstance(scenario_str, str):
                    scenario_data = json.loads(scenario_str)
                    period = scenario_data.get('investment_period', 'medium')
                    sector = scenario_data.get('sector', 'Unknown')
                    trading_scenarios = scenario_data.get('trading_scenarios', {})
                    initial_stop_loss = scenario_data.get('stop_loss', stop_loss)
                    initial_target_price = scenario_data.get('target_price', target_price)

                    if 'highest_price' in scenario_data:
                        highest_price = scenario_data['highest_price']
                    else:
                        highest_price = max(buy_price, current_price)
                        highest_price_initialized = True
                        logger.info(f"{ticker} highest_price not in scenario, initialized to ${highest_price:,.2f}")

                    # Update highest_price if current price exceeds it
                    if current_price > highest_price:
                        highest_price = current_price
                        scenario_data['highest_price'] = highest_price
                        updated_scenario_str = json.dumps(scenario_data, ensure_ascii=False)
                        # Pyramiding (#288): scope by row id so only THIS row's
                        # scenario is updated. Fall back to ticker when id missing.
                        row_id = stock_data.get('id')
                        if row_id is not None:
                            self.cursor.execute(
                                "UPDATE us_stock_holdings SET scenario = ? WHERE id = ?",
                                (updated_scenario_str, row_id)
                            )
                        else:
                            self.cursor.execute(
                                "UPDATE us_stock_holdings SET scenario = ? WHERE ticker = ? AND account_key = ?",
                                (updated_scenario_str, ticker, self._account_scope()[0])
                            )
                        self.conn.commit()
                        logger.info(f"{ticker} highest_price updated in scenario: ${highest_price:,.2f}")
            except Exception:
                pass

            # Hard mechanical stop-loss check BEFORE AI — cannot be overridden
            if stop_loss > 0 and current_price <= stop_loss:
                logger.info(f"{ticker} Mechanical stop-loss triggered (stop-loss: ${stop_loss:,.2f}) — skipping AI")
                return True, f"Stop-loss condition reached (stop-loss: ${stop_loss:,.2f})"

            # Collect current portfolio info from us_stock_holdings
            self.cursor.execute("""
                SELECT ticker, company_name, buy_price, current_price, scenario
                FROM us_stock_holdings
                WHERE account_key = ?
            """, (self._account_scope()[0],))
            holdings = [dict(row) for row in self.cursor.fetchall()]

            sector_distribution = {}
            investment_periods = {"short": 0, "medium": 0, "long": 0}
            for h in holdings:
                try:
                    h_scenario = json.loads(h.get('scenario', '{}')) if isinstance(h.get('scenario'), str) else {}
                    h_sector = h_scenario.get('sector', 'Other')
                    sector_distribution[h_sector] = sector_distribution.get(h_sector, 0) + 1
                    h_period = h_scenario.get('investment_period', 'medium')
                    investment_periods[h_period] = investment_periods.get(h_period, 0) + 1
                except Exception:
                    sector_distribution['Other'] = sector_distribution.get('Other', 0) + 1

            portfolio_info = (
                f"Current Holdings: {len(holdings)}/{self.max_slots}\n"
                f"Sector Distribution: {json.dumps(sector_distribution)}\n"
                f"Investment Period Distribution: {json.dumps(investment_periods)}"
            )

            logger.info(f"[_analyze_sell_decision] {ticker}({company_name}) portfolio_info:")
            logger.info(f"  - Holdings: {len(holdings)}/{self.max_slots}, Sectors: {json.dumps(sector_distribution)}")

            # Fetch portfolio adjustment history for this ticker
            adjustment_history_section = ""
            try:
                self.cursor.execute("""
                    SELECT adjusted_at, old_target_price, new_target_price,
                           old_stop_loss, new_stop_loss, adjustment_reason, urgency
                    FROM us_portfolio_adjustment_log
                    WHERE ticker = ? AND account_key = ?
                    ORDER BY adjusted_at DESC LIMIT 10
                """, (ticker, self._account_scope()[0]))
                adj_rows = self.cursor.fetchall()
                if adj_rows:
                    lines = ["### Portfolio Adjustment History:"]
                    for r in adj_rows:
                        ot = r[1] or 0
                        nt = r[2] or 0
                        os_ = r[3] or 0
                        ns = r[4] or 0
                        reason = r[5] or "N/A"
                        urg = r[6] or "N/A"
                        lines.append(
                            f"- [{r[0][:16]}] Target: ${ot:,.2f}→${nt:,.2f} / "
                            f"Stop: ${os_:,.2f}→${ns:,.2f} ({urg}) — {reason}"
                        )
                    adjustment_history_section = "\n".join(lines)
                    logger.info(f"[_analyze_sell_decision] {ticker} adjustment history: {len(adj_rows)} records injected")
            except Exception:
                pass  # Table may not exist yet on first run

            # Dynamic trailing stop threshold: min 1.5%, max 5%, scales with price appreciation
            trailing_stop_threshold_pct = max(1.5, min(5.0, (highest_price - buy_price) / buy_price * 100 * 0.3)) if buy_price > 0 else 3.0

            # LLM call
            llm = await self.sell_decision_agent.attach_llm(OpenAIAugmentedLLM)

            # 시스템이 사이클당 1회 계산한 LIVE 시장 레짐(S&P500/VIX 기반, OpenAI 무관).
            # 매수시점 동결값(scenario.market_condition)이 아닌 '현재' 레짐을 권위값으로 주입.
            live_regime_summary = getattr(self, "_live_regime_summary", None)
            live_regime_block = (
                live_regime_summary
                if live_regime_summary
                else "unavailable — verify via yahoo_finance (^GSPC 20MA, ^VIX)"
            )

            prompt_message = f"""
Please make a sell/hold decision for the following US stock holding.

### Current Market Regime (LIVE, system-computed — authoritative):
{live_regime_block}

### Stock Information:
- Stock: {company_name} ({ticker})
- Buy Price: ${buy_price:,.2f}
- Current Price: ${current_price:,.2f}
- Target Price: ${target_price:,.2f} (initial scenario: ${initial_target_price:,.2f})
- Stop Loss: ${stop_loss:,.2f} (initial scenario: ${initial_stop_loss:,.2f})
- Highest Price Since Entry: ${highest_price:,.2f}{' (⚠️ First tracking - verify actual peak since entry via get_historical_stock_prices)' if highest_price_initialized else ''}
- Trailing Stop Adjustment Threshold: {trailing_stop_threshold_pct:.1f}% (only adjust stop-loss if new value is at least this much higher)
- Return: {profit_rate:.2f}%
- Holding Period: {days_passed} days
- Investment Period: {period}
- Sector: {sector}

### Current Portfolio Status:
{portfolio_info}

### Trading Scenario:
{json.dumps(trading_scenarios, ensure_ascii=False) if trading_scenarios else "No scenario information"}

{adjustment_history_section}

### Task:
Use yahoo_finance and sqlite tools to check latest data, then decide whether to sell or continue holding.
**Market regime**: Treat the "Current Market Regime (LIVE)" above as the authoritative market environment for step-0 분석 (강세장/약세장 판단). It is system-computed from S&P500/VIX this cycle; prefer it over the stored buy-time scenario. Only if it shows "unavailable", fall back to fetching ^GSPC/^VIX via yahoo_finance yourself.
**Important**: If stop loss/target price adjustment is needed, return it via portfolio_adjustment JSON only. Do NOT directly UPDATE the DB.
"""

            response = await llm.generate_str(
                message=prompt_message,
                request_params=RequestParams(model="gpt-5.5", maxTokens=30000)
            )

            if not response or not response.strip():
                logger.warning(f"{ticker} Empty LLM response, falling back to rule-based decision")
                return await self._fallback_sell_decision(stock_data)

            # Parse JSON from response
            json_str = None
            markdown_match = re.search(r'```(?:json)?\s*({[\s\S]*?})\s*```', response, re.DOTALL)
            if markdown_match:
                json_str = markdown_match.group(1)
            if not json_str:
                json_match = re.search(r'(\{(?:[^{}]|\{(?:[^{}]|\{[^{}]*\})*\})*\})', response, re.DOTALL)
                if json_match:
                    json_str = json_match.group(1)
            if not json_str:
                clean = response.strip()
                if clean.startswith('{') and clean.endswith('}'):
                    json_str = clean

            if not json_str:
                logger.warning(f"{ticker} No JSON found in LLM response, falling back to rule-based decision")
                return await self._fallback_sell_decision(stock_data)

            json_str = re.sub(r',(\s*[}\]])', r'\1', json_str)
            try:
                decision_json = json.loads(json_str)
            except json.JSONDecodeError:
                json_str_clean = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', json_str)
                decision_json = json.loads(json_str_clean)

            should_sell = decision_json.get("should_sell", False)
            sell_reason = decision_json.get("sell_reason", "AI analysis result")
            confidence = decision_json.get("confidence", 5)
            analysis_summary = decision_json.get("analysis_summary", {})
            portfolio_adjustment = decision_json.get("portfolio_adjustment", {})
            logger.info(f"{ticker}({company_name}) AI sell decision: {'Sell' if should_sell else 'Hold'} (confidence: {confidence}/10)")
            logger.info(f"Sell reason: {sell_reason}")

            # Process portfolio_adjustment when holding (not selling)
            if not should_sell and portfolio_adjustment.get("needed", False):
                await self._process_portfolio_adjustment(
                    ticker, company_name, portfolio_adjustment, analysis_summary, current_price,
                    row_id=stock_data.get('id')
                )

            return should_sell, sell_reason

        except Exception as e:
            logger.error(f"{ticker} AI sell analysis error: {e}, falling back to rule-based decision")
            return await self._fallback_sell_decision(stock_data)

    def _get_live_regime_safe(self) -> Optional[str]:
        """매도 사이클의 '현재' 시장 레짐을 yfinance(S&P500/VIX) 기반으로 1회 계산.
        OpenAI 와 무관하므로 quota 사고 중에도 동작. 실패 시 None → stale 폴백.

        부수효과: self._live_regime_summary 에 AI 매도 프롬프트 주입용 사람-읽기 요약을
        저장한다(S&P500 vs 20MA, 4주 변동, VIX). 실패 시 None.
        """
        self._live_regime_summary = None
        try:
            from cores.data_prefetch import prefetch_us_macro_intelligence_data
            data = prefetch_us_macro_intelligence_data() or {}
            cr = data.get("computed_regime") or {}
            regime = cr.get("market_regime")
            if regime:
                logger.info(f"[sell] live US market regime: {regime}")
                s = cr.get("index_summary") or {}
                conf = cr.get("regime_confidence")
                parts = [regime]
                if s.get("sp500_current") is not None and s.get("sp500_20d_ma") is not None:
                    parts.append(
                        f"S&P500 {s['sp500_current']} vs 20MA {s['sp500_20d_ma']} "
                        f"({s.get('sp500_vs_20d_ma','?')})"
                    )
                if s.get("sp500_4w_change_pct") is not None:
                    parts.append(f"4w {s['sp500_4w_change_pct']}%")
                if s.get("vix_current") is not None:
                    parts.append(f"VIX {s['vix_current']} ({s.get('vix_level','?')})")
                if conf is not None:
                    parts.append(f"confidence {conf}")
                self._live_regime_summary = " | ".join(str(p) for p in parts)
            return regime or None
        except Exception as e:
            logger.warning(f"[sell] live regime fetch failed, using stale market_condition: {e}")
            return None

    async def _fallback_sell_decision(self, stock_data: Dict[str, Any]) -> Tuple[bool, str]:
        """Rule-based sell decision (fallback when AI unavailable).

        1차: O'Neil 추세추종 룰(cores.oneil_fallback) — 승자 보유/손실 차단.
        2차(안전망): 모듈 import 실패 등 예외 시에만 기존 레거시 룰.

        Args:
            stock_data: Stock information

        Returns:
            Tuple[bool, str]: Whether to sell, sell reason
        """
        # ── O'Neil 룰베이스 (live regime 주입) ───────────────────
        if _ONEIL_FALLBACK_AVAILABLE:
            try:
                live_regime = getattr(self, "_live_regime_cache", None)
                inp = _oneil_from(stock_data, live_regime=live_regime)
                should_sell, reason = _oneil_eval(inp)
                logger.info(
                    f"{stock_data.get('ticker','')} O'Neil rule-based sell: "
                    f"{'Sell' if should_sell else 'Hold'} | {reason}"
                )
                return should_sell, reason
            except Exception as e:
                logger.error(f"O'Neil fallback error, using legacy rules: {e}")

        # ── 레거시 안전망 (O'Neil 모듈 불가/예외 시에만) ─────────
        try:
            ticker = stock_data.get('ticker', '')
            buy_price = stock_data.get('buy_price', 0)
            buy_date = stock_data.get('buy_date', '')
            current_price = stock_data.get('current_price', 0)
            target_price = stock_data.get('target_price', 0)
            stop_loss = stock_data.get('stop_loss', 0)

            # Calculate profit rate
            profit_rate = ((current_price - buy_price) / buy_price) * 100 if buy_price > 0 else 0

            # Days elapsed from buy date
            buy_datetime = datetime.strptime(buy_date, "%Y-%m-%d %H:%M:%S")
            days_passed = (datetime.now() - buy_datetime).days

            # Extract scenario information
            scenario_str = stock_data.get('scenario', '{}')
            investment_period = "medium"

            try:
                if isinstance(scenario_str, str):
                    scenario_data = json.loads(scenario_str)
                    investment_period = scenario_data.get('investment_period', 'medium')
            except Exception:
                pass

            # Check stop-loss condition (same format as KR template)
            if stop_loss > 0 and current_price <= stop_loss:
                return True, f"Stop-loss condition reached (stop-loss: ${stop_loss:,.2f})"

            # Check target price reached
            if target_price > 0 and current_price >= target_price:
                return True, f"Target price achieved (target: ${target_price:,.2f})"

            # Sell conditions by investment period
            if investment_period == "short":
                # Short-term investment: quicker sell (15+ days holding + 5%+ profit)
                if days_passed >= 15 and profit_rate >= 5:
                    return True, f"Short-term investment goal achieved (holding: {days_passed} days, return: {profit_rate:.2f}%)"
                # Short-term investment loss protection (10+ days + 3%+ loss)
                if days_passed >= 10 and profit_rate <= -3:
                    return True, f"Short-term investment loss protection (holding: {days_passed} days, return: {profit_rate:.2f}%)"

            # General sell conditions
            # Sell if profit >= 10%
            if profit_rate >= 10:
                return True, f"Return exceeds 10% (current return: {profit_rate:.2f}%)"

            # Sell if loss >= 5%
            if profit_rate <= -5:
                return True, f"Loss exceeds -5% (current return: {profit_rate:.2f}%)"

            # Sell if holding 30+ days with loss
            if days_passed >= 30 and profit_rate < 0:
                return True, f"Held 30+ days with loss (holding: {days_passed} days, return: {profit_rate:.2f}%)"

            # Sell if holding 60+ days with 3%+ profit
            if days_passed >= 60 and profit_rate >= 3:
                return True, f"Held 60+ days with 3%+ profit (holding: {days_passed} days, return: {profit_rate:.2f}%)"

            # Long-term investment case (90+ days holding + loss)
            if investment_period == "long" and days_passed >= 90 and profit_rate < 0:
                return True, f"Long-term investment loss cleanup (holding: {days_passed} days, return: {profit_rate:.2f}%)"

            # Continue holding by default
            return False, "Continue holding"

        except Exception as e:
            logger.error(f"Error analyzing sell decision: {str(e)}")
            return False, "Analysis error"

    async def _process_portfolio_adjustment(
        self,
        ticker: str,
        company_name: str,
        portfolio_adjustment: Dict[str, Any],
        analysis_summary: Dict[str, Any],
        current_price: float = 0,
        row_id: int = None
    ):
        """Process DB updates and Telegram notifications based on portfolio_adjustment

        row_id: pyramiding (#288) — when provided, all UPDATEs/SELECT target only
                this specific holding row (multi-row correctness). Falls back to
                ticker-scoped queries when None.
        """
        try:
            if not portfolio_adjustment.get("needed", False):
                return

            urgency = portfolio_adjustment.get("urgency", "low").lower()
            if urgency == "low":
                logger.info(f"{ticker} Portfolio adjustment suggestion (urgency=low): {portfolio_adjustment.get('reason', '')}")
                return

            # Verify holding exists in DB
            if row_id is not None:
                self.cursor.execute(
                    "SELECT target_price, stop_loss FROM us_stock_holdings WHERE id = ?",
                    (row_id,)
                )
            else:
                self.cursor.execute(
                    "SELECT target_price, stop_loss FROM us_stock_holdings WHERE ticker = ? AND account_key = ?",
                    (ticker, self._account_scope()[0])
                )
            row = self.cursor.fetchone()
            if row is None:
                logger.warning(f"{ticker} us_stock_holdings SELECT returned None - skipping adjustment")
                return
            old_target_price = row[0] or 0
            old_stop_loss = row[1] or 0

            db_updated = False
            update_message = ""
            adjustment_reason = portfolio_adjustment.get("reason", "AI analysis result")

            # Adjust target price
            new_target_price = portfolio_adjustment.get("new_target_price")
            if new_target_price is not None:
                try:
                    target_price_num = float(str(new_target_price).replace(',', '').replace('$', ''))
                except (ValueError, TypeError):
                    target_price_num = 0
                if target_price_num > 0:
                    if row_id is not None:
                        self.cursor.execute(
                            "UPDATE us_stock_holdings SET target_price = ? WHERE id = ?",
                            (target_price_num, row_id)
                        )
                    else:
                        self.cursor.execute(
                            "UPDATE us_stock_holdings SET target_price = ? WHERE ticker = ? AND account_key = ?",
                            (target_price_num, ticker, self._account_scope()[0])
                        )
                    self.conn.commit()
                    db_updated = True
                    if target_price_num > old_target_price:
                        direction = "upward"
                    elif target_price_num < old_target_price:
                        direction = "downward"
                    else:
                        direction = "maintained"
                    update_message += f"Target: ${target_price_num:,.2f} ({direction})\n"
                    logger.info(f"{ticker} Target price AI {direction} adjustment: ${target_price_num:,.2f} (prev: ${old_target_price:,.2f})")

            # Adjust stop-loss
            new_stop_loss = portfolio_adjustment.get("new_stop_loss")
            if new_stop_loss is not None:
                try:
                    stop_loss_num = float(str(new_stop_loss).replace(',', '').replace('$', ''))
                except (ValueError, TypeError):
                    stop_loss_num = 0
                if stop_loss_num > 0:
                    # Validation: reject stop_loss above current price
                    if current_price > 0 and stop_loss_num > current_price:
                        logger.warning(
                            f"{ticker} Portfolio adjustment REJECTED: new stop_loss ${stop_loss_num:,.2f} > "
                            f"current_price ${current_price:,.2f}. "
                            f"This indicates trailing stop breach — should trigger sell, not adjustment."
                        )
                    # Ratchet: reject stop_loss below current stop_loss (one-way ratchet)
                    elif old_stop_loss > 0 and stop_loss_num < old_stop_loss:
                        logger.warning(
                            f"{ticker} Ratchet rule violated REJECTED: AI attempted to lower stop_loss "
                            f"${stop_loss_num:,.2f} < ${old_stop_loss:,.2f} — ignoring."
                        )
                    else:
                        if row_id is not None:
                            self.cursor.execute(
                                "UPDATE us_stock_holdings SET stop_loss = ? WHERE id = ?",
                                (stop_loss_num, row_id)
                            )
                        else:
                            self.cursor.execute(
                                "UPDATE us_stock_holdings SET stop_loss = ? WHERE ticker = ? AND account_key = ?",
                                (stop_loss_num, ticker, self._account_scope()[0])
                            )
                        self.conn.commit()
                        db_updated = True
                        if stop_loss_num > old_stop_loss:
                            direction = "upward"
                        elif stop_loss_num < old_stop_loss:
                            direction = "downward"
                        else:
                            direction = "maintained"
                        update_message += f"Stop Loss: ${stop_loss_num:,.2f} ({direction})\n"
                        logger.info(f"{ticker} Stop-loss AI {direction} adjustment: ${stop_loss_num:,.2f} (prev: ${old_stop_loss:,.2f})")

            if db_updated:
                # Log adjustment history (single record for both target + stop_loss changes)
                try:
                    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    acct_key = self._account_scope()[0]
                    # Determine final new values
                    final_new_target = old_target_price
                    try:
                        t = float(str(portfolio_adjustment.get("new_target_price", 0)).replace(',', '').replace('$', ''))
                        if t > 0:
                            final_new_target = t
                    except (ValueError, TypeError):
                        pass
                    final_new_sl = old_stop_loss
                    try:
                        s = float(str(portfolio_adjustment.get("new_stop_loss", 0)).replace(',', '').replace('$', ''))
                        if s > 0:
                            final_new_sl = s
                    except (ValueError, TypeError):
                        pass
                    self.cursor.execute("""
                        INSERT INTO us_portfolio_adjustment_log
                        (account_key, ticker, adjusted_at, old_target_price, new_target_price,
                         old_stop_loss, new_stop_loss, adjustment_reason, urgency)
                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """, (acct_key, ticker, now,
                          old_target_price, final_new_target,
                          old_stop_loss, final_new_sl,
                          adjustment_reason, urgency))
                    self.conn.commit()
                except Exception as log_err:
                    logger.warning(f"{ticker} Failed to log US portfolio adjustment (non-critical): {log_err}")

                urgency_emoji = {"high": "🚨", "medium": "⚠️", "low": "💡"}.get(urgency, "🔄")
                message = f"{urgency_emoji} Portfolio Adjustment: {company_name}({ticker})\n"
                message += update_message
                message += f"Reason: {adjustment_reason}\n"
                message += f"Urgency: {urgency.upper()}\n"
                if analysis_summary:
                    message += f"Technical Trend: {analysis_summary.get('technical_trend', 'N/A')}\n"
                    message += f"Market Impact: {analysis_summary.get('market_condition_impact', 'N/A')}"
                self._msg_types.append("portfolio")
                self.message_queue.append(message)
                logger.info(f"{ticker} AI-based portfolio adjustment complete: {update_message.strip()}")
            else:
                logger.warning(f"{ticker} Portfolio adjustment requested but no specific values: {portfolio_adjustment}")

        except Exception as e:
            logger.error(f"{ticker} Error processing portfolio adjustment: {str(e)}")
            import traceback
            logger.error(traceback.format_exc())

    async def _save_holding_decision(
        self,
        ticker: str,
        current_price: float,
        should_sell: bool,
        sell_reason: str,
        stock_data: Dict[str, Any]
    ) -> bool:
        """
        Save AI sell decision results for held stocks to us_holding_decisions table.
        Main flow continues even if this fails.

        Args:
            ticker: Stock ticker
            current_price: Current price
            should_sell: Whether to sell
            sell_reason: Reason for decision
            stock_data: Full stock data for context

        Returns:
            bool: Save success status
        """
        try:
            now = datetime.now()
            decision_date = now.strftime("%Y-%m-%d")
            decision_time = now.strftime("%H:%M:%S")
            account_key = stock_data.get("account_key") or self._account_scope()[0]
            account_name = stock_data.get("account_name") or self._account_scope()[1]

            # Build decision JSON for storage
            buy_price = stock_data.get('buy_price', 0)
            profit_rate = ((current_price - buy_price) / buy_price * 100) if buy_price > 0 else 0

            decision_json = {
                "should_sell": should_sell,
                "sell_reason": sell_reason,
                "confidence": 7 if should_sell else 5,  # Rule-based confidence
                "analysis_summary": {
                    "technical_trend": "Rule-based analysis",
                    "volume_analysis": "",
                    "market_condition_impact": "",
                    "time_factor": f"Holding days: {stock_data.get('holding_days', 0)}"
                },
                "portfolio_adjustment": {
                    "needed": False,
                    "reason": "",
                    "new_target_price": stock_data.get('target_price'),
                    "new_stop_loss": stock_data.get('stop_loss'),
                    "urgency": "low"
                },
                "current_price": current_price,
                "buy_price": buy_price,
                "profit_rate": profit_rate
            }

            full_json_data = json.dumps(decision_json, ensure_ascii=False)

            # Delete existing data then insert new (keep only latest decision for same ticker)
            self.cursor.execute("DELETE FROM us_holding_decisions WHERE ticker = ? AND account_key = ?", (ticker, account_key))

            # Insert new decision
            self.cursor.execute("""
                INSERT INTO us_holding_decisions (
                    account_key, account_name, ticker, decision_date, decision_time, current_price, should_sell,
                    sell_reason, confidence, technical_trend, volume_analysis,
                    market_condition_impact, time_factor, portfolio_adjustment_needed,
                    adjustment_reason, new_target_price, new_stop_loss, adjustment_urgency,
                    full_json_data
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                account_key, account_name, ticker, decision_date, decision_time, current_price, should_sell,
                sell_reason, decision_json.get("confidence", 5),
                decision_json["analysis_summary"]["technical_trend"],
                decision_json["analysis_summary"]["volume_analysis"],
                decision_json["analysis_summary"]["market_condition_impact"],
                decision_json["analysis_summary"]["time_factor"],
                decision_json["portfolio_adjustment"]["needed"],
                decision_json["portfolio_adjustment"]["reason"],
                decision_json["portfolio_adjustment"]["new_target_price"],
                decision_json["portfolio_adjustment"]["new_stop_loss"],
                decision_json["portfolio_adjustment"]["urgency"],
                full_json_data
            ))

            self.conn.commit()
            logger.info(f"{ticker} US holding decision saved - should_sell: {should_sell}")
            return True

        except Exception as e:
            logger.error(f"{ticker} US holding decision save failed (main flow continues): {str(e)}")
            return False

    async def _delete_holding_decision(self, ticker: str) -> bool:
        """
        Delete decision data for sold stocks from us_holding_decisions table.
        Main flow continues even if this fails.

        Args:
            ticker: Stock ticker

        Returns:
            bool: Delete success status
        """
        try:
            self.cursor.execute("DELETE FROM us_holding_decisions WHERE ticker = ? AND account_key = ?", (ticker, self._account_scope()[0]))
            self.conn.commit()
            logger.info(f"{ticker} US holding decision deleted")
            return True
        except Exception as e:
            logger.error(f"{ticker} US holding decision delete failed: {str(e)}")
            return False

    async def sell_stock(self, stock_data: Dict[str, Any], sell_reason: str,
                         exit_kind: Optional[str] = None) -> bool:
        """
        Process stock sale.

        Args:
            stock_data: Stock information to sell
            sell_reason: Sell reason
            exit_kind: Optional explicit exit classification (stop | trend_exit |
                target | ai). Loops pass it deterministically (loop_a=stop,
                loop_b=trend_exit); when None it is inferred from sell_reason. Stored
                in us_trading_history so the re-entry cooldown treats a stop-out at a
                marginal profit as churn-risk.

        Returns:
            bool: Sell success status
        """
        try:
            ticker = stock_data.get('ticker', '')
            company_name = stock_data.get('company_name', '')
            buy_price = stock_data.get('buy_price', 0)
            buy_date = stock_data.get('buy_date', '')
            current_price = stock_data.get('current_price', 0)
            scenario_json = stock_data.get('scenario', '{}')
            trigger_type = stock_data.get('trigger_type', 'AI_Analysis')
            trigger_mode = stock_data.get('trigger_mode', 'unknown')
            sector = stock_data.get('sector', 'Unknown')
            account_key = stock_data.get("account_key") or self._account_scope()[0]
            account_name = stock_data.get("account_name") or self._account_scope()[1]

            # ── Cross-cycle sell guard (single source of truth) ──────────────
            # EVERY sell path routes its real order + signal publish through
            # sell_stock and gates on this bool return: the batch update_holdings,
            # loop_a_hardstop, and loop_b_trend_exit (KR + US). A concurrent cycle
            # may have already closed this position seconds/minutes ago, so refresh
            # the connection snapshot (commit ends any stale WAL read-txn so other
            # processes' commits are visible) and abort if the row is gone — no
            # trading_history row, no delete, no journal, no queued message — so the
            # caller publishes NO duplicate/ghost SELL and P&L is not double-counted.
            # Incident 2026-07-01 (MU): loop_a stop-sold 23:50 (+published SELL),
            # the batch re-hit the same stop off a stale snapshot and re-published a
            # 2nd SELL 23:55. sell_stock is the chokepoint that closes this for all
            # paths in both markets. (update_holdings also has an earlier Layer 2
            # short-circuit; this is the authoritative gate that also covers loops.)
            self.conn.commit()
            if get_us_existing_position_for_ticker(
                self.cursor, ticker, account_key=account_key
            ).get("row_count", 0) == 0:
                logger.warning(
                    f"[SELL-GUARD][US] {ticker} ({company_name}) already closed by "
                    f"another cycle — sell_stock aborting (no duplicate record/signal)"
                )
                return False
            # ─────────────────────────────────────────────────────────────────

            # Calculate profit rate
            profit_rate = ((current_price - buy_price) / buy_price) * 100 if buy_price > 0 else 0

            # Calculate holding period
            buy_datetime = datetime.strptime(buy_date, "%Y-%m-%d %H:%M:%S")
            holding_days = (datetime.now() - buy_datetime).days
            now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

            # Classify the exit (stop/trend_exit/target/ai) for the churn guard.
            try:
                from reentry_cooldown import classify_exit_kind
                _exit_kind = classify_exit_kind(sell_reason, exit_kind)
            except Exception:
                _exit_kind = exit_kind  # fail-open: store caller hint or None

            # Add to trading history
            self.cursor.execute(
                """
                INSERT INTO us_trading_history
                (account_key, account_name, ticker, company_name, buy_price, buy_date, sell_price, sell_date,
                 profit_rate, holding_days, scenario, trigger_type, trigger_mode, sector, exit_kind)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    account_key, account_name, ticker, company_name, buy_price, buy_date,
                    current_price, now, profit_rate, holding_days,
                    scenario_json, trigger_type, trigger_mode, sector, _exit_kind
                )
            )

            # Remove from holdings.
            # Pyramiding (#288): when the row carries an id AND the ticker has more
            # than one row for this account, delete ONLY that row (preserving the
            # remaining independent entries). Single-row tickers keep the legacy
            # ticker-scoped delete + adjustment-log cleanup (zero behavior change).
            row_id = stock_data.get('id')
            existing = get_us_existing_position_for_ticker(self.cursor, ticker, account_key=account_key)
            is_partial = row_id is not None and existing.get("row_count", 0) > 1
            if is_partial:
                self.cursor.execute(
                    "DELETE FROM us_stock_holdings WHERE id = ?",
                    (row_id,)
                )
            else:
                self.cursor.execute(
                    "DELETE FROM us_stock_holdings WHERE ticker = ? AND account_key = ?",
                    (ticker, account_key)
                )
                # Cleanup portfolio adjustment history only when the ticker is fully
                # exited (no remaining rows).
                try:
                    self.cursor.execute(
                        "DELETE FROM us_portfolio_adjustment_log WHERE ticker = ? AND account_key = ?",
                        (ticker, account_key)
                    )
                except Exception as e:
                    logger.debug(f"{ticker} Cleanup adjustment log skipped: {e}")
            self.conn.commit()

            # Build sell message (same format as KR template)
            arrow = "⬆️" if profit_rate > 0 else "⬇️" if profit_rate < 0 else "➖"
            message = f"📉 Sell: {company_name}({ticker})\n" \
                      f"Buy Price: ${buy_price:,.2f}\n" \
                      f"Sell Price: ${current_price:,.2f}\n" \
                      f"Return: {arrow} {abs(profit_rate):.2f}%\n" \
                      f"Holding Period: {holding_days} days\n" \
                      f"Sell Reason: {sell_reason}"

            # Add trigger win rate
            trigger_type = stock_data.get('trigger_type', '')
            trigger_win_rate = self._get_trigger_win_rate(trigger_type)
            if trigger_win_rate:
                message += f"\n{trigger_win_rate}"

            self._msg_types.append("analysis")
            self.message_queue.append(message)
            logger.info(f"{ticker} ({company_name}) sell complete (return: {profit_rate:.2f}%)")

            # Create trading journal entry (if enabled)
            if self.enable_journal and self.journal_manager:
                try:
                    await self.journal_manager.create_entry(
                        stock_data=stock_data,
                        sell_price=current_price,
                        profit_rate=profit_rate,
                        holding_days=holding_days,
                        sell_reason=sell_reason
                    )
                    logger.info(f"US Journal entry created for {ticker}")
                except Exception as journal_err:
                    logger.warning(f"Failed to create US journal entry: {journal_err}")

            return True

        except Exception as e:
            logger.error(f"Error during sell: {str(e)}")
            logger.error(traceback.format_exc())
            return False

    async def update_holdings(self) -> List[Dict[str, Any]]:
        """
        Update holdings information and make sell decisions.

        Returns:
            List[Dict]: List of sold stock information
        """
        try:
            logger.info("Starting US holdings update")

            # 매도 판단에 쓸 '현재' 시장 레짐을 사이클당 1회 계산(OpenAI 무관).
            # _fallback_sell_decision 이 self._live_regime_cache 로 참조한다.
            self._live_regime_cache = self._get_live_regime_safe()

            # Query holdings list
            # id included for pyramiding (#288): enables per-row delete and
            # fractional-sell quantity computation for multi-row tickers.
            self.cursor.execute(
                """SELECT id, ticker, company_name, buy_price, buy_date, current_price,
                   scenario, target_price, stop_loss, last_updated,
                   trigger_type, trigger_mode, sector, account_key, account_name
                   FROM us_stock_holdings
                   WHERE account_key = ?""",
                (self._account_scope()[0],)
            )
            holdings = [dict(row) for row in self.cursor.fetchall()]

            if not holdings:
                logger.info("No US holdings")
                return []

            sold_stocks = []

            # Pyramiding (#288) FIX 2 — in-pass over-sell guard (mirror of KR):
            # snapshot the ticker's total broker qty ONCE per pass and distribute
            # from (snapshot - already_ordered), so unfilled limit/reserved orders
            # cannot cause an over-sell on later iterations of the same ticker.
            pass_total_qty: Dict[str, int] = {}   # ticker -> snapshot total qty
            pass_sold_qty: Dict[str, int] = {}    # ticker -> cumulative ordered qty
            # FIX 1 — tickers already FULL-exited this pass (all rows sold at once
            # because the order had to be queued). Their remaining DB rows were
            # already removed, so skip them when the loop reaches them.
            fully_exited_tickers: set = set()

            for stock in holdings:
                ticker = stock.get('ticker')
                company_name = stock.get('company_name')

                # FIX 1: skip rows whose ticker was already fully exited this pass.
                if ticker in fully_exited_tickers:
                    logger.info(f"{ticker} already fully exited this pass — skipping remaining row")
                    continue

                # Query current stock price
                current_price = await self._get_current_stock_price(ticker)

                if current_price <= 0:
                    old_price = stock.get('current_price', 0)
                    logger.warning(f"{ticker} current price query failed, using last: ${old_price:.2f}")
                    current_price = old_price

                stock['current_price'] = current_price
                now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

                # Analyze sell decision
                should_sell, sell_reason = await self._analyze_sell_decision(stock)

                if should_sell:
                    acct_key = stock.get("account_key")

                    # ── Layer 2: fresh holding re-check (cross-cycle stale-snapshot guard) ──
                    # update_holdings iterates a holdings snapshot taken at pipeline
                    # start (~40 min earlier). A concurrent intraday loop
                    # (loop_a_hardstop / loop_b_trend_exit, both */10 on cron) may have
                    # already closed this position minutes ago. Re-read the live DB row
                    # right before acting; if it is gone, abort THIS ticker entirely —
                    # no real order, NO signal publish, no journal — so subscribers never
                    # receive a duplicate/ghost SELL.
                    # Incident 2026-07-01: loop_a stop-sold MU 23:50 (+published SELL),
                    # then the batch re-hit the same stop off its stale snapshot 23:54,
                    # its real sell no-op'd ("not found in portfolio") yet it still
                    # published a 2nd SELL 23:55. This guard closes that gap at the source.
                    self.conn.commit()  # end any read snapshot so other-process commits are visible (WAL)
                    if get_us_existing_position_for_ticker(
                        self.cursor, ticker, account_key=acct_key
                    ).get("row_count", 0) == 0:
                        logger.warning(
                            f"[LAYER2][US] {ticker} already closed by another cycle "
                            f"(stale snapshot) — skipping sell + signal publish"
                        )
                        continue
                    # ── end Layer 2 guard ──

                    # Pyramiding (#288): compute remaining row count N for this
                    # (ticker, account) BEFORE any DB row is deleted by sell_stock.
                    remaining_rows = get_us_existing_position_for_ticker(
                        self.cursor, ticker, account_key=acct_key
                    ).get("row_count", 1)

                    # FIX 1: decide the execution plan. A fractional (N>1) order
                    # that would be QUEUED for the pending-order batch must NOT be
                    # sent as a partial (the queue drops the quantity and later
                    # full-liquidates the broker position). In that case do a FULL
                    # position exit instead: sell the whole holding once and delete
                    # ALL rows for the ticker so DB and broker stay consistent.
                    will_queue = False
                    if remaining_rows > 1 and current_price > 0:
                        try:
                            try:
                                from trading.us_stock_trading import AsyncUSTradingContext
                            except ImportError:
                                from prism_us.trading.us_stock_trading import AsyncUSTradingContext
                            async with AsyncUSTradingContext(account_name=stock.get("account_name")) as _probe:
                                # Order gets queued when market is closed AND the
                                # reserved-order window is unavailable (pre-10:00 KST).
                                will_queue = (not _probe.is_market_open()) and (not _probe.is_reserved_order_available())
                        except Exception as probe_err:
                            logger.warning(f"{ticker} could not probe order window ({probe_err}); assuming will_queue=True (safe full-exit)")
                            will_queue = True

                    plan = decide_us_sell_plan(remaining_rows, will_queue)
                    logger.info(f"{ticker} sell plan: {plan} (remaining_rows={remaining_rows}, will_queue={will_queue})")

                    # Delete from holding_decisions when selling
                    await self._delete_holding_decision(ticker)

                    if plan == "full_exit":
                        # Record P&L for EVERY remaining row at current price and
                        # delete them all (sell_stock deletes one row per call;
                        # the final call deletes the last row by ticker scope).
                        self.cursor.execute(
                            """SELECT id, ticker, company_name, buy_price, buy_date, current_price,
                               scenario, target_price, stop_loss, last_updated,
                               trigger_type, trigger_mode, sector, account_key, account_name
                               FROM us_stock_holdings
                               WHERE ticker = ? AND account_key = ?""",
                            (ticker, acct_key),
                        )
                        sibling_rows = [dict(r) for r in self.cursor.fetchall()]
                        sell_success = True
                        for sib in sibling_rows:
                            sib['current_price'] = current_price
                            ok = await self.sell_stock(sib, sell_reason)
                            sell_success = sell_success and ok
                        # Mark immediately (independent of KIS result) so the
                        # already-deleted sibling rows still in the in-memory
                        # `holdings` snapshot are skipped later this pass.
                        fully_exited_tickers.add(ticker)
                        logger.info(
                            f"{ticker} FULL-EXIT: closed {len(sibling_rows)} pyramid rows "
                            f"(queued order cannot carry a partial quantity)"
                        )
                    else:
                        sell_success = await self.sell_stock(stock, sell_reason)

                    if sell_success:
                        # Execute actual trading
                        trade_result = {'success': False, 'message': 'Trading not executed'}

                        # Only execute trading if we have a valid price
                        if current_price > 0:
                            try:
                                try:
                                    from trading.us_stock_trading import AsyncUSTradingContext
                                except ImportError:
                                    from prism_us.trading.us_stock_trading import AsyncUSTradingContext
                                async with AsyncUSTradingContext(account_name=stock.get("account_name")) as trading:
                                    # Determine sell quantity.
                                    # FIX 1: full_exit -> quantity=None (sell whole position).
                                    # FIX 2: fractional -> distribute from a per-pass snapshot
                                    # (snapshot - already_ordered) so unfilled orders cannot over-sell.
                                    sell_quantity = None
                                    # The FINAL row of a ticker already split THIS pass
                                    # (plan flips to "single_full" at remaining_rows==1) must also
                                    # sell from the snapshot remainder, NOT re-query the broker —
                                    # otherwise unfilled earlier limit orders make get_holding_quantity
                                    # return the full position and the last row over-sells (#288 FIX 2).
                                    if plan == "fractional" or (ticker in pass_total_qty and plan == "single_full"):
                                        if ticker not in pass_total_qty:
                                            pass_total_qty[ticker] = await asyncio.to_thread(
                                                trading.get_holding_quantity, ticker
                                            )
                                            pass_sold_qty[ticker] = 0
                                        available = pass_total_qty[ticker] - pass_sold_qty[ticker]
                                        sell_quantity = compute_us_fractional_sell_quantity(available, remaining_rows)
                                        pass_sold_qty[ticker] += sell_quantity
                                        logger.info(
                                            f"{ticker} pyramiding fractional sell: {sell_quantity} shares "
                                            f"(available {available} of snapshot {pass_total_qty[ticker]}, "
                                            f"remaining rows={remaining_rows})"
                                        )
                                    # plan == "full_exit" -> sell_quantity stays None (full position);
                                    # the ticker was already added to fully_exited_tickers above.
                                    # Pass limit_price for reserved orders (required for US market)
                                    # If limit_price is 0, trading module will use MOO (Market On Open)
                                    trade_result = await trading.async_sell_stock(ticker=ticker, limit_price=current_price, quantity=sell_quantity)

                                if trade_result['success']:
                                    logger.info(f"Actual sell successful: {trade_result['message']}")
                                else:
                                    logger.error(f"Actual sell failed: {trade_result['message']}")
                            except Exception as trade_err:
                                logger.warning(f"Trading execution skipped: {trade_err}")
                        else:
                            logger.warning(f"Skipping actual sell for {ticker}: invalid current_price ({current_price})")

                        # [Optional] Publish sell signal via Redis Streams
                        # Auto-skipped if Redis not configured (requires UPSTASH_REDIS_REST_URL, UPSTASH_REDIS_REST_TOKEN)
                        try:
                            from messaging.redis_signal_publisher import publish_sell_signal
                            profit_rate = ((current_price - stock.get('buy_price', 0)) / stock.get('buy_price', 0) * 100) if stock.get('buy_price', 0) > 0 else 0
                            await publish_sell_signal(
                                ticker=ticker,
                                company_name=company_name,
                                price=current_price,
                                buy_price=stock.get('buy_price', 0),
                                profit_rate=profit_rate,
                                sell_reason=sell_reason,
                                trade_result=trade_result,
                                market="US"
                            )
                        except Exception as signal_err:
                            logger.warning(f"Sell signal publish failed (non-critical): {signal_err}")

                        # [Optional] Publish sell signal via GCP Pub/Sub
                        # Auto-skipped if GCP not configured (requires GCP_PROJECT_ID, GCP_PUBSUB_TOPIC_ID)
                        try:
                            from messaging.gcp_pubsub_signal_publisher import publish_sell_signal as gcp_publish_sell_signal
                            profit_rate = ((current_price - stock.get('buy_price', 0)) / stock.get('buy_price', 0) * 100) if stock.get('buy_price', 0) > 0 else 0
                            await gcp_publish_sell_signal(
                                ticker=ticker,
                                company_name=company_name,
                                price=current_price,
                                buy_price=stock.get('buy_price', 0),
                                profit_rate=profit_rate,
                                sell_reason=sell_reason,
                                trade_result=trade_result,
                                market="US"
                            )
                        except Exception as signal_err:
                            logger.warning(f"GCP sell signal publish failed (non-critical): {signal_err}")

                        # Report sold rows. For a full_exit, every closed pyramid
                        # row is reported (so the sell count matches reality);
                        # otherwise just the single row that triggered the sell.
                        _exited_rows = sibling_rows if plan == "full_exit" else [stock]
                        for _row in _exited_rows:
                            _bp = _row.get('buy_price', 0)
                            sold_stocks.append({
                                "ticker": ticker,
                                "company_name": _row.get('company_name', company_name),
                                "buy_price": _bp,
                                "sell_price": current_price,
                                "profit_rate": ((current_price - _bp) / _bp * 100) if _bp > 0 else 0,
                                "reason": sell_reason,
                                "account_name": _row.get("account_name"),
                                "account_label": self._safe_account_log_label(
                                    {
                                        "name": _row.get("account_name"),
                                        "account_key": _row.get("account_key"),
                                    }
                                ),
                            })
                else:
                    # Save holding decision when not selling
                    await self._save_holding_decision(ticker, current_price, should_sell, sell_reason, stock)

                    # Update current price
                    self.cursor.execute(
                        """UPDATE us_stock_holdings
                           SET current_price = ?, last_updated = ?
                           WHERE ticker = ? AND account_key = ?""",
                        (current_price, now, ticker, stock.get("account_key"))
                    )
                    self.conn.commit()
                    logger.info(f"{ticker} ({company_name}) price updated: ${current_price:.2f} ({sell_reason})")

            return sold_stocks

        except Exception as e:
            logger.error(f"Error updating holdings: {str(e)}")
            logger.error(traceback.format_exc())
            return []

    async def generate_report_summary(self) -> str:
        """
        Generate holdings and profit statistics summary.

        Returns:
            str: Summary message
        """
        try:
            # Query holdings
            self.cursor.execute(
                """SELECT ticker, company_name, buy_price, current_price, buy_date,
                   scenario, target_price, stop_loss, sector
                   FROM us_stock_holdings
                   WHERE account_key = ?""",
                (self._account_scope()[0],)
            )
            holdings = [dict(row) for row in self.cursor.fetchall()]

            # Calculate total profit from trading history
            self.cursor.execute("SELECT SUM(profit_rate) FROM us_trading_history WHERE account_key = ?", (self._account_scope()[0],))
            total_profit = self.cursor.fetchone()[0] or 0

            # Number of trades
            self.cursor.execute("SELECT COUNT(*) FROM us_trading_history WHERE account_key = ?", (self._account_scope()[0],))
            total_trades = self.cursor.fetchone()[0] or 0

            # Number of successful trades
            self.cursor.execute("SELECT COUNT(*) FROM us_trading_history WHERE account_key = ? AND profit_rate > 0", (self._account_scope()[0],))
            successful_trades = self.cursor.fetchone()[0] or 0

            # Generate message (Korean as default - same as Korean stock version)
            message = f"📊 프리즘 US 시뮬레이터 | 실시간 포트폴리오 ({datetime.now().strftime('%Y-%m-%d %H:%M')})\n\n"

            # 1. Portfolio summary
            message += f"🔸 Current Holdings: {len(holdings) if holdings else 0}/{self.max_slots}\n"

            # Best profit/loss stock information (if any)
            if holdings and len(holdings) > 0:
                profit_rates = []
                for h in holdings:
                    buy_price = h.get('buy_price', 0)
                    current_price = h.get('current_price', 0)
                    if buy_price > 0:
                        profit_rate = ((current_price - buy_price) / buy_price) * 100
                        profit_rates.append((h.get('ticker'), h.get('company_name'), profit_rate))

                if profit_rates:
                    best = max(profit_rates, key=lambda x: x[2])
                    worst = min(profit_rates, key=lambda x: x[2])

                    message += f"✅ 최고 수익: {best[1]}({best[0]}) {'+' if best[2] > 0 else ''}{best[2]:.2f}%\n"
                    message += f"⚠️ 최저 수익: {worst[1]}({worst[0]}) {'+' if worst[2] > 0 else ''}{worst[2]:.2f}%\n"

            message += "\n"

            # 2. Sector distribution analysis
            sector_counts = {}

            if holdings and len(holdings) > 0:
                message += "🔸 Holdings List:\n"
                for stock in holdings:
                    ticker = stock.get('ticker', '')
                    company_name = stock.get('company_name', '')
                    buy_price = stock.get('buy_price', 0)
                    current_price = stock.get('current_price', 0)
                    buy_date = stock.get('buy_date', '')
                    target_price = stock.get('target_price', 0)
                    stop_loss = stock.get('stop_loss', 0)
                    scenario_str = stock.get('scenario', '{}')

                    # Extract sector information from scenario
                    sector = "알 수 없음"
                    try:
                        if isinstance(scenario_str, str):
                            scenario_data = json.loads(scenario_str)
                            sector = scenario_data.get('sector', '알 수 없음')
                    except Exception:
                        sector = stock.get('sector', '알 수 없음')

                    # Update sector count
                    sector_counts[sector] = sector_counts.get(sector, 0) + 1

                    profit_rate = ((current_price - buy_price) / buy_price) * 100 if buy_price > 0 else 0
                    arrow = "⬆️" if profit_rate > 0 else "⬇️" if profit_rate < 0 else "➖"

                    buy_datetime = datetime.strptime(buy_date, "%Y-%m-%d %H:%M:%S") if buy_date else datetime.now()
                    days_passed = (datetime.now() - buy_datetime).days

                    message += f"- {company_name}({ticker}) [{sector}]\n"
                    message += f"  Buy: ${buy_price:.2f} / Current: ${current_price:.2f}\n"
                    message += f"  Target: ${target_price:.2f} / Stop: ${stop_loss:.2f}\n"
                    message += f"  수익률: {arrow} {profit_rate:.2f}% / 보유기간: {days_passed}일\n\n"

                # Add sector distribution
                message += "🔸 Sector Distribution:\n"
                for sector, count in sector_counts.items():
                    percentage = (count / len(holdings)) * 100
                    message += f"- {sector}: {count}개 ({percentage:.1f}%)\n"
                message += "\n"
            else:
                message += "No holdings.\n\n"

            # 3. Trading history statistics
            message += "🔸 매매 이력 통계\n"
            message += f"- 총 거래 건수: {total_trades}건\n"
            message += f"- 수익 거래: {successful_trades}건\n"
            message += f"- 손실 거래: {total_trades - successful_trades}건\n"

            if total_trades > 0:
                message += f"- 승률: {(successful_trades / total_trades * 100):.2f}%\n"
            else:
                message += "- 승률: 0.00%\n"

            message += f"- 누적 수익률: {total_profit:.2f}%\n\n"

            # 4. Enhanced Disclaimer
            message += "📝 Important Notice:\n"
            message += "- This report is an AI-based simulation result and is not related to actual trading.\n"
            message += "- This information is for reference only. Investment decisions and responsibilities lie solely with the investor.\n"
            message += "- This channel is not a trading room and does not recommend buying/selling specific stocks."

            return message

        except Exception as e:
            logger.error(f"Error generating report summary: {str(e)}")
            return f"Error generating report: {str(e)}"

    async def process_reports(self, pdf_report_paths: List[str]) -> Tuple[int, int]:
        """
        Process analysis reports and make buy/sell decisions.

        Args:
            pdf_report_paths: List of PDF analysis report file paths

        Returns:
            Tuple[int, int]: Buy count, sell count
        """
        try:
            logger.info(f"Processing {len(pdf_report_paths)} US reports")

            if not self.account_configs:
                logger.warning("No accounts configured. Skipping buy/sell execution.")
                return 0, 0

            if not self.active_account:
                self._set_active_account(self.account_configs[0])

            buy_count = 0
            sell_count = 0
            signaled_tickers: set[str] = set()
            analysis_states: list[dict[str, Any]] = []

            for pdf_report_path in pdf_report_paths:
                analysis_result = await self._analyze_report_core(pdf_report_path)
                if not analysis_result.get("success", False):
                    logger.error(
                        f"[ANALYSIS_FAILED] Report analysis skipped "
                        f"({analysis_result.get('ticker', '?')}/{analysis_result.get('company_name', '?')}): "
                        f"{analysis_result.get('error')} — {pdf_report_path}"
                    )
                    continue
                analysis_states.append(
                    {
                        "analysis": analysis_result,
                        "traded": False,
                        "should_save_watchlist": False,
                        "skip_reason": None,
                    }
                )

            for account in self.account_configs:
                self._set_active_account(account)
                label = self._safe_account_log_label(account)
                logger.info(f"Processing US reports for account {label}")

                # Update existing holdings and make sell decisions
                sold_stocks = await self.update_holdings()
                sell_count += len(sold_stocks)

                if sold_stocks:
                    logger.info(f"{len(sold_stocks)} stocks sold for {label}")
                else:
                    logger.info(f"No stocks sold for {label}")

                for state in analysis_states:
                    analysis_result = state["analysis"]
                    ticker = analysis_result.get("ticker")
                    company_name = analysis_result.get("company_name")
                    current_price = analysis_result.get("current_price", 0)
                    scenario = analysis_result.get("scenario", {})
                    sector = analysis_result.get("sector", "Unknown")
                    rank_change_msg = analysis_result.get("rank_change_msg", "")

                    # Pyramiding (#288): allow an additional independent entry for a
                    # held ticker only when the strong-bull add-gate passes. Otherwise
                    # keep the legacy skip. Computed per active account.
                    is_add = False
                    if await self._is_ticker_in_holdings(ticker):
                        acct_key = self._account_scope()[0]
                        existing = get_us_existing_position_for_ticker(self.cursor, ticker, account_key=acct_key)
                        allowed, gate_reason = evaluate_us_pyramid_add_gate(
                            market_condition=scenario.get("market_condition", ""),
                            existing_avg_buy_price=existing.get("avg_buy_price", 0.0),
                            current_price=current_price,
                            existing_row_count=existing.get("row_count", 0),
                        )
                        if not allowed:
                            logger.info(f"Skipping stock in holdings: {ticker} — add gate blocked: {gate_reason}")
                            continue
                        logger.info(f"{ticker} pyramiding add gate passed: {gate_reason}")
                        is_add = True

                    current_slots = await self._get_current_slots_count()
                    if current_slots >= self.max_slots:
                        # User-facing reason must NOT include the account label (leaks
                        # masked account number to broadcast channels). Keep detail in logs only.
                        reason = "Max slots reached (portfolio full)"
                        logger.info(f"Purchase deferred: {company_name} ({ticker}) - Max slots reached for {label}")
                        state["should_save_watchlist"] = True
                        state["skip_reason"] = state["skip_reason"] or reason
                        continue

                    # Evaluate sector / score / decision independently so the displayed
                    # rejection reason matches the rationale in the same message,
                    # rather than short-circuiting on whichever cause the code checked first.
                    sector_diverse = await self._check_sector_diversity(sector, is_pyramiding_add=is_add)

                    buy_score = scenario.get("buy_score", 0)
                    min_score = scenario.get("min_score", 0)

                    score_adjustment = 0
                    adjustment_reasons = []
                    trigger_info = getattr(self, 'trigger_info_map', {}).get(ticker, {})
                    trigger_type = trigger_info.get('trigger_type', '')
                    if self.enable_journal and ticker:
                        score_adjustment, adjustment_reasons = self.get_score_adjustment(ticker, sector, trigger_type=trigger_type)
                        if score_adjustment != 0:
                            logger.info(
                                f"Journal score adjustment for {ticker}: {score_adjustment:+d} "
                                f"(reasons: {', '.join(adjustment_reasons)})"
                            )

                    adjusted_score = buy_score + score_adjustment
                    logger.info(
                        f"Buy score: {company_name} ({ticker}) - Original: {buy_score}, "
                        f"Adjusted: {adjusted_score}, Min: {min_score}"
                    )

                    raw_decision = analysis_result.get("raw_decision", "")
                    normalized_decision = analysis_result.get("decision", "no_entry")
                    if raw_decision and raw_decision.lower() != normalized_decision:
                        logger.debug(f"Decision normalized: '{raw_decision}' -> '{normalized_decision}'")

                    rationale = scenario.get("rationale", "") or ""
                    logger.info(
                        f"Scenario decision: {company_name} ({ticker}) - "
                        f"decision={normalized_decision!r}, sector_diverse={sector_diverse}, sector={sector!r}"
                    )
                    if rationale:
                        logger.info(f"Scenario rationale ({company_name}/{ticker}): {rationale[:300]}")

                    # Re-entry cooldown gate (SHADOW logs only; LIVE vetoes a churn
                    # re-entry — longer cooldown after a loss). Fresh entries only;
                    # pyramiding adds (is_add) are exempt.
                    _cd_block = False
                    if normalized_decision == "entry" and not is_add:
                        try:
                            from reentry_cooldown import reentry_block, COOLDOWN_LIVE, COOLDOWN_RISK_EXIT_LIVE
                            _cd = reentry_block("US", ticker)
                        except Exception:
                            _cd, COOLDOWN_LIVE, COOLDOWN_RISK_EXIT_LIVE = None, False, False
                        if _cd:
                            # A stop/trend-exit block that is NOT also a loss is the new
                            # exit-kind branch -> SHADOW unless COOLDOWN_RISK_EXIT_LIVE.
                            _risk_only = bool(_cd.get("risk_exit")) and not _cd.get("after_loss")
                            _enforce = COOLDOWN_LIVE and (COOLDOWN_RISK_EXIT_LIVE or not _risk_only)
                            logger.warning(
                                "[REENTRY_COOLDOWN][%s] %s ticker=%s last_sell=%s ret=%.1f%% gap=%.1fh<%sh after_loss=%s exit_kind=%s risk_only=%s",
                                "LIVE" if _enforce else "SHADOW", _cd["action"], ticker,
                                _cd["last_sell"], _cd["last_ret"], _cd["gap_hours"],
                                _cd["window_hours"], _cd["after_loss"], _cd.get("exit_kind"), _risk_only)
                            _cd_block = _enforce

                    if normalized_decision == "entry" and adjusted_score >= min_score and sector_diverse and not _cd_block:
                        # is_add => pyramiding additional independent row (#288)
                        buy_success = await self.buy_stock(ticker, company_name, current_price, scenario, rank_change_msg, is_add=is_add)

                        if buy_success:
                            trade_result = {'success': False, 'message': 'Trading not executed'}

                            if current_price > 0:
                                try:
                                    try:
                                        from trading.us_stock_trading import AsyncUSTradingContext
                                    except ImportError:
                                        from prism_us.trading.us_stock_trading import AsyncUSTradingContext
                                    async with AsyncUSTradingContext(account_name=account["name"]) as trading:
                                        trade_result = await trading.async_buy_stock(ticker=ticker, limit_price=current_price)

                                    if trade_result['success']:
                                        logger.info(f"Actual purchase successful: {trade_result['message']}")
                                    else:
                                        logger.error(f"Actual purchase failed: {trade_result['message']}")
                                except Exception as trade_err:
                                    logger.warning(f"Trading execution skipped: {trade_err}")
                            else:
                                logger.warning(f"Skipping actual purchase for {ticker}: invalid current_price ({current_price})")

                            # Simulator DB record (inserted by buy_stock) is independent of KIS result.
                            # KIS failure only affects real-money execution — the simulator holding stays.
                            trade_actually_succeeded = trade_result.get('success') or trade_result.get('partial_success')

                            # Simulator state: always update when buy_stock() succeeded,
                            # regardless of KIS result (simulator and real trading are independent).
                            buy_count += 1
                            state["traded"] = True

                            if not trade_actually_succeeded:
                                logger.warning(
                                    f"[{ticker}] KIS order failed: {trade_result.get('message', 'Unknown')} "
                                    f"— simulator holding preserved, no skip notification"
                                )
                                logger.info(f"Simulator purchase recorded: {company_name} ({ticker}) @ ${current_price:.2f} (KIS order failed)")
                            else:
                                if trade_result.get("partial_success"):
                                    successful = trade_result.get("successful_accounts", [])
                                    failed = trade_result.get("failed_accounts", [])
                                    logger.warning(
                                        f"{ticker} partial success: {len(successful)}/{len(successful) + len(failed)} accounts"
                                    )

                                if ticker not in signaled_tickers:
                                    try:
                                        from messaging.redis_signal_publisher import publish_buy_signal
                                        await publish_buy_signal(
                                            ticker=ticker,
                                            company_name=company_name,
                                            price=current_price,
                                            scenario=scenario,
                                            source="AI분석",
                                            trade_result=trade_result,
                                            market="US"
                                        )
                                    except Exception as signal_err:
                                        logger.warning(f"Buy signal publish failed (non-critical): {signal_err}")

                                    try:
                                        from messaging.gcp_pubsub_signal_publisher import publish_buy_signal as gcp_publish_buy_signal
                                        await gcp_publish_buy_signal(
                                            ticker=ticker,
                                            company_name=company_name,
                                            price=current_price,
                                            scenario=scenario,
                                            source="AI분석",
                                            trade_result=trade_result,
                                            market="US"
                                        )
                                    except Exception as signal_err:
                                        logger.warning(f"GCP buy signal publish failed (non-critical): {signal_err}")

                                    signaled_tickers.add(ticker)

                                logger.info(f"Purchase complete: {company_name} ({ticker}) @ ${current_price:.2f}")
                        else:
                            state["should_save_watchlist"] = True
                            state["skip_reason"] = state["skip_reason"] or "Purchase failed"
                            logger.warning(f"Purchase failed: {company_name} ({ticker})")
                    else:
                        # Build a single reason string that lists ALL applicable causes,
                        # so the displayed reason matches the AI rationale shown in the
                        # same message (instead of short-circuiting on sector check first).
                        reason_parts = []
                        if normalized_decision != "entry":
                            reason_parts.append(f"AI judgment: {normalized_decision}")
                        if adjusted_score < min_score:
                            reason_parts.append(f"Insufficient score ({adjusted_score}/{min_score})")
                        if not sector_diverse:
                            reason_parts.append(f"Sector concentration ({sector})")
                        reason = " / ".join(reason_parts) if reason_parts else "Other"
                        logger.info(f"Purchase deferred: {company_name} ({ticker}) - {reason}")
                        state["should_save_watchlist"] = True
                        state["skip_reason"] = state["skip_reason"] or reason

            for state in analysis_states:
                if state["traded"] or not state["should_save_watchlist"]:
                    continue

                analysis_result = state["analysis"]
                scenario = analysis_result.get("scenario", {})
                decision = analysis_result.get("decision", "no_entry")
                await self._save_watchlist_item(
                    ticker=analysis_result.get("ticker"),
                    company_name=analysis_result.get("company_name"),
                    current_price=analysis_result.get("current_price", 0),
                    buy_score=scenario.get("buy_score", 0),
                    min_score=scenario.get("min_score", 0),
                    decision=decision if decision != "entry" else "Skip",
                    skip_reason=state["skip_reason"] or "Trade not executed",
                    scenario=scenario,
                    sector=analysis_result.get("sector", "Unknown"),
                    was_traded=False
                )

            logger.info(f"Report processing complete - Bought: {buy_count}, Sold: {sell_count}")
            return buy_count, sell_count

        except Exception as e:
            logger.error(f"Error processing reports: {str(e)}")
            logger.error(traceback.format_exc())
            return 0, 0

    async def _notify_firebase(self, message: str, chat_id: str, message_id: int = None, msg_type=None):
        """Send Firebase Bridge notification for Prism Mobile push (never affects Telegram delivery)."""
        try:
            from firebase_bridge import notify
            await notify(
                message=message,
                market="us",
                telegram_message_id=message_id,
                channel_id=chat_id,
                msg_type=msg_type,
            )
        except Exception as e:
            logger.debug(f"Firebase bridge: {e}")

    def _schedule_firebase(self, message: str, chat_id: str, message_id: int = None, msg_type=None):
        """Schedule Firebase notification as non-blocking task. Returns the task."""
        return asyncio.create_task(self._notify_firebase(message, chat_id, message_id, msg_type=msg_type))

    async def send_telegram_message(self, chat_id: str, language: str = "ko") -> bool:
        """
        Send message via Telegram.

        Args:
            chat_id: Telegram channel ID (no sending if None)
            language: Message language ("ko" or "en")

        Returns:
            bool: Send success status
        """
        try:
            # Skip Telegram sending if chat_id is None
            if not chat_id:
                logger.info("No Telegram channel ID. Skipping message send")

                # Log message output
                for message in self.message_queue:
                    logger.info(f"[Message (not sent)] {message[:100]}...")

                # Initialize message queue
                self.message_queue = []
                self._msg_types = []
                return True  # Consider intentional skip as success

            # If Telegram bot not initialized, only output logs
            if not self.telegram_bot:
                logger.warning("Telegram bot not initialized. Please check token")

                # Only output messages without actual sending
                for message in self.message_queue:
                    logger.info(f"[Telegram message (bot not initialized)] {message[:100]}...")

                # Initialize message queue
                self.message_queue = []
                self._msg_types = []
                return False

            # Generate summary report — de-duplicated so near-simultaneous run-ends
            # (US batch + intraday loops A/B) don't emit 2-3 identical portfolio
            # summaries. Other queued messages (sell notices) are unaffected.
            try:
                from portfolio_broadcast import should_send_portfolio
                _emit_portfolio = should_send_portfolio("US")
            except Exception:
                _emit_portfolio = True  # fail-open
            if _emit_portfolio:
                summary = await self.generate_report_summary()
                self._msg_types.append("portfolio")
                self.message_queue.append(summary)
            else:
                logger.info("[portfolio-dedup] US portfolio summary skipped (sent within debounce window)")

            # Translate messages if English is requested
            if language == "en":
                logger.info(f"Translating {len(self.message_queue)} US messages to English")
                try:
                    # Note: translate_telegram_message is pre-loaded at module level
                    # from main project's cores/agents/telegram_translator_agent.py
                    translated_queue = []
                    for idx, message in enumerate(self.message_queue, 1):
                        logger.info(f"Translating US message {idx}/{len(self.message_queue)}")
                        translated = await translate_telegram_message(message, model="gpt-5.4-nano")
                        translated_queue.append(translated)
                    self.message_queue = translated_queue
                    logger.info("All US messages translated successfully")
                except Exception as e:
                    logger.error(f"Translation failed: {str(e)}. Using original Korean messages.")

            # Send each message (Firebase notifications are non-blocking)
            success = True
            firebase_tasks = []
            for idx, message in enumerate(self.message_queue):
                msg_type = self._msg_types[idx] if idx < len(self._msg_types) else None
                logger.info(f"Sending US Telegram message: {chat_id}")
                try:
                    # Telegram message length limit (4096 characters)
                    MAX_MESSAGE_LENGTH = 4096

                    if len(message) <= MAX_MESSAGE_LENGTH:
                        # Send short message at once
                        result = await self.telegram_bot.send_message(
                            chat_id=chat_id,
                            text=message
                        )
                        firebase_tasks.append(self._schedule_firebase(message, chat_id, result.message_id, msg_type=msg_type))
                    else:
                        # Split long message
                        parts = []
                        current_part = ""

                        for line in message.split('\n'):
                            if len(current_part) + len(line) + 1 <= MAX_MESSAGE_LENGTH:
                                current_part += line + '\n'
                            else:
                                if current_part:
                                    parts.append(current_part.rstrip())
                                current_part = line + '\n'

                        if current_part:
                            parts.append(current_part.rstrip())

                        # Send split messages
                        first_msg_id = None
                        for i, part in enumerate(parts, 1):
                            result = await self.telegram_bot.send_message(
                                chat_id=chat_id,
                                text=f"[{i}/{len(parts)}]\n{part}"
                            )
                            if i == 1:
                                first_msg_id = result.message_id
                            await asyncio.sleep(0.5)  # Short delay between split messages

                        # Notify with full original message, link to first part
                        firebase_tasks.append(self._schedule_firebase(message, chat_id, first_msg_id, msg_type=msg_type))

                    logger.info(f"US Telegram message sent: {chat_id}")
                except TelegramError as e:
                    logger.error(f"US Telegram message send failed: {e}")
                    success = False

                # Delay to prevent API rate limiting
                await asyncio.sleep(1)

            # Gather Firebase notifications (non-blocking for Telegram delivery)
            if firebase_tasks:
                await asyncio.gather(*firebase_tasks, return_exceptions=True)

            # Send to broadcast channels if configured (awaited in run() finally block)
            if hasattr(self, 'telegram_config') and self.telegram_config and self.telegram_config.broadcast_languages:
                self._broadcast_task = asyncio.create_task(self._send_to_translation_channels(self.message_queue.copy(), self._msg_types.copy()))
                logger.info("US broadcast channel translation dispatched")

            # Clear message queue
            self.message_queue = []
            self._msg_types = []

            return success

        except Exception as e:
            logger.error(f"Error sending US Telegram message: {str(e)}")
            logger.error(traceback.format_exc())
            return False

    async def _send_to_translation_channels(self, messages: List[str], msg_types: Optional[list] = None):
        """
        Send messages to translation channels

        Args:
            messages: List of original Korean messages
            msg_types: msg_type for each message in the list
        """
        try:
            # Note: translate_telegram_message is pre-loaded at module level
            # from main project's cores/agents/telegram_translator_agent.py

            for lang in self.telegram_config.broadcast_languages:
                try:
                    # Get channel ID for this language
                    channel_id = self.telegram_config.get_broadcast_channel_id(lang)
                    if not channel_id:
                        logger.warning(f"No channel ID configured for language: {lang}")
                        continue

                    logger.info(f"Sending US tracking messages to {lang} channel")

                    # Translate and send each message (Firebase non-blocking)
                    firebase_tasks = []
                    for msg_idx, message in enumerate(messages):
                        msg_type = msg_types[msg_idx] if msg_types and msg_idx < len(msg_types) else None
                        try:
                            # Translate message
                            logger.info(f"Translating US tracking message to {lang}")
                            translated_message = await translate_telegram_message(
                                message,
                                model="gpt-5.4-nano",
                                from_lang="ko",
                                to_lang=lang
                            )

                            # Send translated message
                            MAX_MESSAGE_LENGTH = 4096

                            if len(translated_message) <= MAX_MESSAGE_LENGTH:
                                result = await self.telegram_bot.send_message(
                                    chat_id=channel_id,
                                    text=translated_message
                                )
                                firebase_tasks.append(self._schedule_firebase(translated_message, channel_id, result.message_id, msg_type=msg_type))
                            else:
                                # Split long messages
                                parts = []
                                current_part = ""

                                for line in translated_message.split('\n'):
                                    if len(current_part) + len(line) + 1 <= MAX_MESSAGE_LENGTH:
                                        current_part += line + '\n'
                                    else:
                                        if current_part:
                                            parts.append(current_part.rstrip())
                                        current_part = line + '\n'

                                if current_part:
                                    parts.append(current_part.rstrip())

                                first_msg_id = None
                                for i, part in enumerate(parts, 1):
                                    result = await self.telegram_bot.send_message(
                                        chat_id=channel_id,
                                        text=f"[{i}/{len(parts)}]\n{part}"
                                    )
                                    if i == 1:
                                        first_msg_id = result.message_id
                                    await asyncio.sleep(0.5)

                                firebase_tasks.append(self._schedule_firebase(translated_message, channel_id, first_msg_id, msg_type=msg_type))

                            logger.info(f"US tracking message sent successfully to {lang} channel")

                            await asyncio.sleep(1)

                        except Exception as e:
                            logger.error(f"Error translating/sending US message to {lang}: {str(e)}")
                            from telegram_config import is_openai_quota_error, send_openai_quota_alert
                            if is_openai_quota_error(e):
                                await send_openai_quota_alert(self.telegram_config, market="US")
                                return

                    # Gather Firebase notifications for this language
                    if firebase_tasks:
                        await asyncio.gather(*firebase_tasks, return_exceptions=True)

                except Exception as e:
                    logger.error(f"Error processing language {lang}: {str(e)}")

        except Exception as e:
            logger.error(f"Error in _send_to_translation_channels: {str(e)}")

    def get_compression_stats(self) -> Dict[str, Any]:
        """
        Get current compression statistics for US market.

        Returns:
            Dict with compression layer counts and stats
        """
        if self.compression_manager:
            return self.compression_manager.get_compression_stats()
        return {"error": "Compression manager not initialized"}

    async def compress_old_journal_entries(
        self,
        layer1_age_days: int = 7,
        layer2_age_days: int = 30,
        min_entries_for_compression: int = 3
    ) -> Dict[str, Any]:
        """
        Compress old journal entries for US market.

        Args:
            layer1_age_days: Days before Layer 1 entries are compressed
            layer2_age_days: Days before Layer 2 entries are compressed
            min_entries_for_compression: Minimum entries to trigger compression

        Returns:
            Dict with compression results
        """
        if self.compression_manager:
            return await self.compression_manager.compress_old_journal_entries(
                layer1_age_days=layer1_age_days,
                layer2_age_days=layer2_age_days,
                min_entries_for_compression=min_entries_for_compression
            )
        return {"error": "Compression manager not initialized"}

    def cleanup_stale_data(
        self,
        max_principles: int = 50,
        max_intuitions: int = 50,
        stale_days: int = 90,
        archive_layer3_days: int = 365,
        dry_run: bool = False
    ) -> Dict[str, Any]:
        """
        Clean up stale data for US market.

        Args:
            max_principles: Maximum active principles to keep
            max_intuitions: Maximum active intuitions to keep
            stale_days: Days without validation before deactivation
            archive_layer3_days: Days after which to archive Layer 3 entries
            dry_run: If True, only count what would be cleaned

        Returns:
            Dict with cleanup results
        """
        if self.compression_manager:
            return self.compression_manager.cleanup_stale_data(
                max_principles=max_principles,
                max_intuitions=max_intuitions,
                stale_days=stale_days,
                archive_layer3_days=archive_layer3_days,
                dry_run=dry_run
            )
        return {"error": "Compression manager not initialized"}

    def get_journal_context(self, ticker: str, sector: str = None, trigger_type: str = None) -> str:
        """
        Get trading journal context for buy decisions.

        Args:
            ticker: Stock ticker symbol
            sector: Stock sector (optional)
            trigger_type: Trigger type for performance tracker lookup (optional)

        Returns:
            str: Context string with past trading experiences
        """
        if self.journal_manager and self.enable_journal:
            return self.journal_manager.get_context_for_ticker(ticker, sector, trigger_type=trigger_type)
        return ""

    def get_score_adjustment(self, ticker: str, sector: str = None, trigger_type: str = None) -> Tuple[int, List[str]]:
        """
        Calculate score adjustment based on past experiences and performance tracker data.

        Args:
            ticker: Stock ticker symbol
            sector: Stock sector (optional)
            trigger_type: Trigger type for performance tracker lookup (optional)

        Returns:
            Tuple[int, List[str]]: Adjustment value (-3 to +3) and reasons
        """
        if self.journal_manager and self.enable_journal:
            return self.journal_manager.get_score_adjustment(ticker, sector, trigger_type=trigger_type)
        return 0, []

    def _get_trigger_win_rate(self, trigger_type: str) -> str:
        """Get trigger win rate string from us_analysis_performance_tracker.
        Returns a formatted string like '(Trigger Win Rate: 63%)' or empty string if no data."""
        if not trigger_type or not self.conn:
            return ""
        try:
            cursor = self.conn.cursor()
            # Check table exists
            table_check = cursor.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='us_analysis_performance_tracker'"
            ).fetchone()
            if not table_check:
                return ""
            row = cursor.execute("""
                SELECT COUNT(*) as completed,
                       SUM(CASE WHEN return_30d > 0 THEN 1 ELSE 0 END) as wins
                FROM us_analysis_performance_tracker
                WHERE trigger_type = ? AND return_30d IS NOT NULL
            """, (trigger_type,)).fetchone()
            if row and row[0] >= 3:
                win_rate = int(row[1] / row[0] * 100)
                return f"📡 Trigger Win Rate: {win_rate}% ({row[0]} trades)"
            return ""
        except Exception:
            return ""

    async def run(self, pdf_report_paths: List[str], chat_id: str = None,
                  language: str = "ko", telegram_config=None, trigger_results_file: str = None,
                  sector_names: list = None) -> bool:
        """
        Main execution function for US stock tracking system.

        Args:
            pdf_report_paths: List of analysis report file paths
            chat_id: Telegram channel ID (optional)
            language: Message language (default: "ko")
            telegram_config: TelegramConfig object for multi-language support
            trigger_results_file: Path to trigger results JSON file

        Returns:
            bool: Execution success status
        """
        try:
            logger.info("Starting US tracking system batch execution")

            # Store telegram_config for use in send_telegram_message
            self.telegram_config = telegram_config

            # Load trigger type mapping
            self.trigger_info_map = {}
            if trigger_results_file:
                try:
                    if os.path.exists(trigger_results_file):
                        with open(trigger_results_file, 'r', encoding='utf-8') as f:
                            trigger_data = json.load(f)
                        for trigger_type, stocks in trigger_data.items():
                            if trigger_type == 'metadata':
                                self.trigger_mode = trigger_data.get('metadata', {}).get('trigger_mode', '')
                                continue
                            if isinstance(stocks, list):
                                for stock in stocks:
                                    ticker = stock.get('ticker', stock.get('code', ''))
                                    if ticker:
                                        self.trigger_info_map[ticker] = {
                                            'trigger_type': trigger_type,
                                            'trigger_mode': trigger_data.get('metadata', {}).get('trigger_mode', ''),
                                            'risk_reward_ratio': stock.get('risk_reward_ratio', 0)
                                        }
                        logger.info(f"Loaded trigger info for {len(self.trigger_info_map)} stocks")
                except Exception as e:
                    logger.warning(f"Failed to load trigger results file: {e}")

            # Initialize
            await self.initialize(language, sector_names=sector_names)

            try:
                # Process reports
                buy_count, sell_count = await self.process_reports(pdf_report_paths)

                # Send Telegram message
                if chat_id:
                    message_sent = await self.send_telegram_message(chat_id, language)
                    if message_sent:
                        logger.info("US Telegram message sent successfully")
                    else:
                        logger.warning("US Telegram message send failed")
                else:
                    logger.info("Telegram channel ID not provided, skipping message send")
                    await self.send_telegram_message(None, language)

                logger.info("US tracking system batch execution complete")
                return True
            finally:
                # Wait for broadcast translation task before cleanup
                if self._broadcast_task:
                    try:
                        logger.info("Waiting for US tracking broadcast translation to complete...")
                        await self._broadcast_task
                        logger.info("US tracking broadcast translation completed")
                    except Exception as e:
                        logger.error(f"US tracking broadcast translation failed: {e}")
                    self._broadcast_task = None

                # Ensure connection is always closed
                if self.conn:
                    self.conn.close()
                    logger.info("Database connection closed")

        except Exception as e:
            logger.error(f"Error during US tracking system execution: {str(e)}")
            logger.error(traceback.format_exc())

            if hasattr(self, 'conn') and self.conn:
                try:
                    self.conn.close()
                except Exception:
                    pass

            return False


async def main():
    """Main function"""
    import argparse

    parser = argparse.ArgumentParser(description="US Stock tracking and trading agent")
    parser.add_argument("--reports", nargs="+", help="List of analysis report file paths")
    parser.add_argument("--chat-id", help="Telegram channel ID")
    parser.add_argument("--telegram-token", help="Telegram bot token")
    parser.add_argument("--language", default="ko", help="Language (default: ko)")
    parser.add_argument(
        "--enable-journal",
        action="store_true",
        help="Enable trading journal for retrospective analysis"
    )

    args = parser.parse_args()

    if not args.reports:
        logger.error("Report path not specified")
        return False

    async with app.run():
        agent = USStockTrackingAgent(
            telegram_token=args.telegram_token,
            enable_journal=args.enable_journal
        )
        success = await agent.run(args.reports, args.chat_id, args.language)
        return success


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        logger.error(f"Error during program execution: {str(e)}")
        logger.error(traceback.format_exc())
        sys.exit(1)
