DinQuant/backend_api_python/app/routes/portfolio.py

"""
Portfolio API routes (local-only).
Manages manual positions (user's existing holdings) and AI monitoring tasks.
"""
from flask import Blueprint, request, jsonify, g
from datetime import date, datetime, timezone
import json
import traceback
import time
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed

from app.services.kline import KlineService
from app.utils.logger import get_logger
from app.utils.cache import CacheManager
from app.utils.db import get_db_connection
from app.utils.auth import login_required
from app.services.symbol_name import resolve_symbol_name
from app.data.market_symbols_seed import get_symbol_name as seed_get_symbol_name

logger = get_logger(__name__)

portfolio_bp = Blueprint('portfolio', __name__)
kline_service = KlineService()
cache = CacheManager()

# Thread pool for parallel price fetching
# Lower concurrency to avoid triggering API limits (especially for forex/US stocks)
executor = ThreadPoolExecutor(max_workers=3)

# Request interval (seconds) to avoid too frequent requests
REQUEST_INTERVAL = 0.3

# Rate limiting related
_request_lock = threading.Lock()
_last_request_time = {}  # {market: timestamp}


def _now_ts() -> int:
    return int(time.time())


def _serialize_monitor_ts(value):
    """
    JSON serialized monitoring time field. PostgreSQL TIMESTAMP is interpreted in UTC when there is no tz (consistent with Docker's default),
    Output the ISO with Z to prevent the front end from treating the non-time zone string as local time and deviating by 8 hours.
    """
    if value is None:
        return None
    if isinstance(value, datetime):
        if value.tzinfo is None:
            dt = value.replace(tzinfo=timezone.utc)
        else:
            dt = value.astimezone(timezone.utc)
        s = dt.isoformat()
        if s.endswith("+00:00"):
            return s[:-6] + "Z"
        return s
    if isinstance(value, date):
        return value.isoformat()
    return value


def _normalize_symbol(symbol: str) -> str:
    return (symbol or '').strip().upper()


def _safe_json_loads(value, default=None):
    """Safely parse JSON string."""
    if default is None:
        default = {}
    if isinstance(value, dict):
        return value
    if isinstance(value, list):
        return value
    if isinstance(value, str) and value.strip():
        try:
            return json.loads(value)
        except Exception:
            return default
    return default


def _get_single_price(market: str, symbol: str, force_refresh: bool = False) -> dict:
    """
    Get price data for a single symbol.
    
    Priority is given to using the real-time quotation API (ticker), and downgrading to minute/daily K-line data.
    This allows for more real-time prices during the trading session, rather than just showing daily closing prices.
    
    Built-in rate limit: requests for the same market must be at least REQUEST_INTERVAL seconds apart,
    Avoid triggering API limits (like yfinance, Tiingo, Finnhub, etc.).
    
    Args:
        force_refresh: whether to force refresh (skip cache)
    """
    try:
        # Rate Limit: Interval between requests for the same market
        with _request_lock:
            now = time.time()
            last_time = _last_request_time.get(market, 0)
            wait_time = REQUEST_INTERVAL - (now - last_time)
            if wait_time > 0:
                time.sleep(wait_time)
            _last_request_time[market] = time.time()
        
        # Get real-time prices using the new get_realtime_price method
        price_data = kline_service.get_realtime_price(market, symbol, force_refresh=force_refresh)
        
        return {
            'market': market,
            'symbol': symbol,
            'price': price_data.get('price', 0),
            'change': price_data.get('change', 0),
            'changePercent': price_data.get('changePercent', 0),
            'source': price_data.get('source', 'unknown')  # Record data sources for easy debugging
        }
    except Exception as e:
        logger.error(f"Failed to fetch price {market}:{symbol} - {str(e)}")
        return {
            'market': market,
            'symbol': symbol,
            'price': 0,
            'change': 0,
            'changePercent': 0,
            'source': 'error'
        }


# ==================== Position CRUD ====================

@portfolio_bp.route('/positions', methods=['GET'])
@login_required
def get_positions():
    """Get all manual positions with current prices for the current user."""
    try:
        user_id = g.user_id
        # Check if force refresh (skip cache)
        force_refresh = request.args.get('refresh', '').lower() in ('1', 'true', 'yes')
        
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                """
                SELECT id, market, symbol, name, side, quantity, entry_price, entry_time, notes, tags, group_name, created_at, updated_at
                FROM qd_manual_positions
                WHERE user_id = ?
                ORDER BY id DESC
                """,
                (user_id,)
            )
            rows = cur.fetchall() or []
            cur.close()

        positions = []
        price_futures = {}
        
        # Prepare positions and submit price fetch tasks
        for row in rows:
            pos = {
                'id': row.get('id'),
                'market': row.get('market'),
                'symbol': row.get('symbol'),
                'name': row.get('name') or row.get('symbol'),
                'side': row.get('side') or 'long',
                'quantity': float(row.get('quantity') or 0),
                'entry_price': float(row.get('entry_price') or 0),
                'entry_time': row.get('entry_time'),
                'notes': row.get('notes') or '',
                'tags': _safe_json_loads(row.get('tags'), []),
                'group_name': row.get('group_name') or '',
                'created_at': row.get('created_at'),
                'updated_at': row.get('updated_at'),
                # Will be filled later
                'current_price': 0,
                'price_change': 0,
                'price_change_percent': 0,
                'market_value': 0,
                'cost_value': 0,
                'pnl': 0,
                'pnl_percent': 0
            }
            positions.append(pos)
            
            # Submit price fetch task (with force_refresh support)
            market = row.get('market')
            symbol = row.get('symbol')
            if market and symbol:
                key = f"{market}:{symbol}"
                if key not in price_futures:
                    future = executor.submit(_get_single_price, market, symbol, force_refresh)
                    price_futures[key] = future

        # Collect price results
        price_map = {}
        for key, future in price_futures.items():
            try:
                result = future.result(timeout=10)
                price_map[key] = result
            except Exception as e:
                logger.warning(f"Price fetch failed for {key}: {e}")

        # Calculate PnL for each position
        for pos in positions:
            key = f"{pos['market']}:{pos['symbol']}"
            price_data = price_map.get(key, {})
            
            current_price = float(price_data.get('price') or 0)
            entry_price = pos['entry_price']
            quantity = pos['quantity']
            side = pos['side']
            
            pos['current_price'] = current_price
            pos['price_change'] = price_data.get('change', 0)
            pos['price_change_percent'] = price_data.get('changePercent', 0)
            
            # Calculate values
            pos['market_value'] = current_price * quantity
            pos['cost_value'] = entry_price * quantity
            
            # Calculate PnL based on side
            if side == 'long':
                pos['pnl'] = (current_price - entry_price) * quantity
            else:  # short
                pos['pnl'] = (entry_price - current_price) * quantity
            
            if pos['cost_value'] > 0:
                pos['pnl_percent'] = round(pos['pnl'] / pos['cost_value'] * 100, 2)

        return jsonify({'code': 1, 'msg': 'success', 'data': positions})
    except Exception as e:
        logger.error(f"get_positions failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': []}), 500


@portfolio_bp.route('/positions', methods=['POST'])
@login_required
def add_position():
    """Add a new manual position for the current user."""
    try:
        user_id = g.user_id
        data = request.get_json() or {}
        market = (data.get('market') or '').strip()
        symbol = _normalize_symbol(data.get('symbol'))
        name_in = (data.get('name') or '').strip()
        side = (data.get('side') or 'long').strip().lower()
        quantity = float(data.get('quantity') or 0)
        entry_price = float(data.get('entry_price') or 0)
        entry_time = data.get('entry_time') or _now_ts()
        notes = (data.get('notes') or '').strip()
        tags = data.get('tags') or []
        group_name = (data.get('group_name') or '').strip()
        
        if not market or not symbol:
            return jsonify({'code': 0, 'msg': 'Missing market or symbol', 'data': None}), 400
        
        if quantity <= 0:
            return jsonify({'code': 0, 'msg': 'Quantity must be positive', 'data': None}), 400
        
        if entry_price <= 0:
            return jsonify({'code': 0, 'msg': 'Entry price must be positive', 'data': None}), 400
        
        if side not in ('long', 'short'):
            side = 'long'
        
        # Resolve display name
        resolved = resolve_symbol_name(market, symbol) or seed_get_symbol_name(market, symbol)
        name = name_in or resolved or symbol
        
        tags_json = json.dumps(tags if isinstance(tags, list) else [], ensure_ascii=False)
        
        with get_db_connection() as db:
            cur = db.cursor()
            # Delete any existing positions for this symbol (regardless of side),
            # ensuring only one position per symbol per user per group.
            cur.execute(
                "DELETE FROM qd_manual_positions WHERE user_id = ? AND market = ? AND symbol = ? AND group_name = ?",
                (user_id, market, symbol, group_name)
            )
            cur.execute(
                """
                INSERT INTO qd_manual_positions 
                (user_id, market, symbol, name, side, quantity, entry_price, entry_time, notes, tags, group_name, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW(), NOW())
                """,
                (user_id, market, symbol, name, side, quantity, entry_price, entry_time, notes, tags_json, group_name)
            )
            position_id = cur.lastrowid
            db.commit()
            cur.close()

        return jsonify({'code': 1, 'msg': 'success', 'data': {'id': position_id}})
    except Exception as e:
        logger.error(f"add_position failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


@portfolio_bp.route('/positions/<int:position_id>', methods=['PUT'])
@login_required
def update_position(position_id):
    """Update an existing position for the current user."""
    try:
        user_id = g.user_id
        data = request.get_json() or {}
        
        updates = []
        params = []
        
        if 'name' in data:
            updates.append('name = ?')
            params.append((data.get('name') or '').strip())
        
        if 'quantity' in data:
            quantity = float(data.get('quantity') or 0)
            if quantity <= 0:
                return jsonify({'code': 0, 'msg': 'Quantity must be positive', 'data': None}), 400
            updates.append('quantity = ?')
            params.append(quantity)
        
        if 'entry_price' in data:
            entry_price = float(data.get('entry_price') or 0)
            if entry_price <= 0:
                return jsonify({'code': 0, 'msg': 'Entry price must be positive', 'data': None}), 400
            updates.append('entry_price = ?')
            params.append(entry_price)
        
        if 'entry_time' in data:
            updates.append('entry_time = ?')
            params.append(data.get('entry_time'))
        
        if 'notes' in data:
            updates.append('notes = ?')
            params.append((data.get('notes') or '').strip())
        
        if 'tags' in data:
            tags = data.get('tags') or []
            updates.append('tags = ?')
            params.append(json.dumps(tags if isinstance(tags, list) else [], ensure_ascii=False))
        
        if 'group_name' in data:
            updates.append('group_name = ?')
            params.append((data.get('group_name') or '').strip())
        
        if not updates:
            return jsonify({'code': 0, 'msg': 'No fields to update', 'data': None}), 400
        
        updates.append('updated_at = NOW()')
        params.append(position_id)
        params.append(user_id)
        
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                f"UPDATE qd_manual_positions SET {', '.join(updates)} WHERE id = ? AND user_id = ?",
                params
            )
            db.commit()
            cur.close()

        return jsonify({'code': 1, 'msg': 'success', 'data': None})
    except Exception as e:
        logger.error(f"update_position failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


@portfolio_bp.route('/positions/<int:position_id>', methods=['DELETE'])
@login_required
def delete_position(position_id):
    """Delete a position for the current user."""
    try:
        user_id = g.user_id
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                "DELETE FROM qd_manual_positions WHERE id = ? AND user_id = ?",
                (position_id, user_id)
            )
            db.commit()
            cur.close()

        return jsonify({'code': 1, 'msg': 'success', 'data': None})
    except Exception as e:
        logger.error(f"delete_position failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


@portfolio_bp.route('/summary', methods=['GET'])
@login_required
def get_portfolio_summary():
    """Get portfolio summary with total value, PnL, and market distribution for the current user."""
    try:
        user_id = g.user_id
        # Check if force refresh
        force_refresh = request.args.get('refresh', '').lower() in ('1', 'true', 'yes')
        
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                """
                SELECT id, market, symbol, side, quantity, entry_price
                FROM qd_manual_positions
                WHERE user_id = ?
                """,
                (user_id,)
            )
            rows = cur.fetchall() or []
            cur.close()

        if not rows:
            return jsonify({
                'code': 1,
                'msg': 'success',
                'data': {
                    'total_cost': 0,
                    'total_market_value': 0,
                    'total_pnl': 0,
                    'total_pnl_percent': 0,
                    'position_count': 0,
                    'market_distribution': []
                }
            })

        # Fetch prices in parallel (with force_refresh support)
        price_futures = {}
        for row in rows:
            market = row.get('market')
            symbol = row.get('symbol')
            key = f"{market}:{symbol}"
            if key not in price_futures:
                future = executor.submit(_get_single_price, market, symbol, force_refresh)
                price_futures[key] = future

        price_map = {}
        for key, future in price_futures.items():
            try:
                result = future.result(timeout=10)
                price_map[key] = result
            except Exception:
                pass

        # Calculate totals
        total_cost = 0
        total_market_value = 0
        total_pnl = 0
        market_values = {}  # {market: market_value}
        
        for row in rows:
            market = row.get('market')
            symbol = row.get('symbol')
            side = row.get('side') or 'long'
            quantity = float(row.get('quantity') or 0)
            entry_price = float(row.get('entry_price') or 0)
            
            key = f"{market}:{symbol}"
            price_data = price_map.get(key, {})
            current_price = float(price_data.get('price') or 0)
            
            cost = entry_price * quantity
            market_val = current_price * quantity
            
            if side == 'long':
                pnl = (current_price - entry_price) * quantity
            else:
                pnl = (entry_price - current_price) * quantity
            
            total_cost += cost
            total_market_value += market_val
            total_pnl += pnl
            
            # Market distribution
            if market not in market_values:
                market_values[market] = 0
            market_values[market] += market_val

        total_pnl_percent = round(total_pnl / total_cost * 100, 2) if total_cost > 0 else 0
        
        # Build market distribution
        market_distribution = []
        for market, value in market_values.items():
            percent = round(value / total_market_value * 100, 2) if total_market_value > 0 else 0
            market_distribution.append({
                'market': market,
                'value': round(value, 2),
                'percent': percent
            })
        
        market_distribution.sort(key=lambda x: x['value'], reverse=True)

        return jsonify({
            'code': 1,
            'msg': 'success',
            'data': {
                'total_cost': round(total_cost, 2),
                'total_market_value': round(total_market_value, 2),
                'total_pnl': round(total_pnl, 2),
                'total_pnl_percent': total_pnl_percent,
                'position_count': len(rows),
                'market_distribution': market_distribution
            }
        })
    except Exception as e:
        logger.error(f"get_portfolio_summary failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


# ==================== Monitor CRUD ====================

@portfolio_bp.route('/monitors', methods=['GET'])
@login_required
def get_monitors():
    """Get all position monitors for the current user."""
    try:
        user_id = g.user_id
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                """
                SELECT id, name, position_ids, monitor_type, config, notification_config, 
                       is_active, last_run_at, next_run_at, last_result, run_count, created_at, updated_at
                FROM qd_position_monitors
                WHERE user_id = ?
                ORDER BY id DESC
                """,
                (user_id,)
            )
            rows = cur.fetchall() or []
            cur.close()

        monitors = []
        for row in rows:
            monitors.append({
                'id': row.get('id'),
                'name': row.get('name') or '',
                'position_ids': _safe_json_loads(row.get('position_ids'), []),
                'monitor_type': row.get('monitor_type') or 'ai',
                'config': _safe_json_loads(row.get('config'), {}),
                'notification_config': _safe_json_loads(row.get('notification_config'), {}),
                'is_active': bool(row.get('is_active')),
                'last_run_at': _serialize_monitor_ts(row.get('last_run_at')),
                'next_run_at': _serialize_monitor_ts(row.get('next_run_at')),
                'last_result': _safe_json_loads(row.get('last_result'), {}),
                'run_count': row.get('run_count') or 0,
                'created_at': _serialize_monitor_ts(row.get('created_at')),
                'updated_at': _serialize_monitor_ts(row.get('updated_at'))
            })

        return jsonify({'code': 1, 'msg': 'success', 'data': monitors})
    except Exception as e:
        logger.error(f"get_monitors failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': []}), 500


@portfolio_bp.route('/monitors', methods=['POST'])
@login_required
def add_monitor():
    """Add a new position monitor for the current user."""
    try:
        user_id = g.user_id
        data = request.get_json() or {}
        name = (data.get('name') or '').strip()
        position_ids = data.get('position_ids') or []
        monitor_type = (data.get('monitor_type') or 'ai').strip()
        config = data.get('config') or {}
        notification_config = data.get('notification_config') or {}
        is_active = bool(data.get('is_active', True))
        
        if not name:
            return jsonify({'code': 0, 'msg': 'Monitor name is required', 'data': None}), 400
        
        if monitor_type not in ('ai', 'price_alert', 'pnl_alert'):
            monitor_type = 'ai'
        
        # Calculate next_run_at based on interval (frontend sends run_interval_minutes)
        interval_minutes = int(config.get('run_interval_minutes') or config.get('interval_minutes') or 60)
        
        position_ids_json = json.dumps(position_ids if isinstance(position_ids, list) else [], ensure_ascii=False)
        config_json = json.dumps(config if isinstance(config, dict) else {}, ensure_ascii=False)
        notification_config_json = json.dumps(notification_config if isinstance(notification_config, dict) else {}, ensure_ascii=False)
        
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                """
                INSERT INTO qd_position_monitors 
                (user_id, name, position_ids, monitor_type, config, notification_config, is_active, next_run_at, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, NOW() + INTERVAL '%s minutes', NOW(), NOW())
                """,
                (user_id, name, position_ids_json, monitor_type, config_json, notification_config_json, 
                 1 if is_active else 0, interval_minutes)
            )
            monitor_id = cur.lastrowid
            db.commit()
            cur.close()

        # Run one round in the background immediately after creation: send a notification immediately, and write next_run_at based on the completion time (execute again after the interval)
        if is_active and monitor_id:
            try:
                from app.services.portfolio_monitor import run_single_monitor as _run_single_monitor

                def _initial_run():
                    try:
                        _run_single_monitor(int(monitor_id), user_id=int(user_id))
                    except Exception as ex:
                        logger.error(f"Initial portfolio monitor run failed #{monitor_id}: {ex}")

                threading.Thread(
                    target=_initial_run,
                    daemon=True,
                    name=f"monitor-init-{monitor_id}",
                ).start()
            except Exception as ex:
                logger.error(f"Failed to schedule initial monitor run #{monitor_id}: {ex}")

        return jsonify({'code': 1, 'msg': 'success', 'data': {'id': monitor_id}})
    except Exception as e:
        logger.error(f"add_monitor failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


@portfolio_bp.route('/monitors/<int:monitor_id>', methods=['PUT'])
@login_required
def update_monitor(monitor_id):
    """Update an existing monitor for the current user."""
    try:
        user_id = g.user_id
        data = request.get_json() or {}
        
        updates = []
        params = []
        
        if 'name' in data:
            updates.append('name = ?')
            params.append((data.get('name') or '').strip())
        
        if 'position_ids' in data:
            position_ids = data.get('position_ids') or []
            updates.append('position_ids = ?')
            params.append(json.dumps(position_ids if isinstance(position_ids, list) else [], ensure_ascii=False))
        
        if 'monitor_type' in data:
            updates.append('monitor_type = ?')
            params.append((data.get('monitor_type') or 'ai').strip())
        
        next_run_interval = None  # Will store interval for special handling
        if 'config' in data:
            config = data.get('config') or {}
            updates.append('config = ?')
            params.append(json.dumps(config if isinstance(config, dict) else {}, ensure_ascii=False))
            
            # Recalculate next_run_at if interval changed
            next_run_interval = int(config.get('run_interval_minutes') or config.get('interval_minutes') or 60)
        
        if 'notification_config' in data:
            notification_config = data.get('notification_config') or {}
            updates.append('notification_config = ?')
            params.append(json.dumps(notification_config if isinstance(notification_config, dict) else {}, ensure_ascii=False))
        
        if 'is_active' in data:
            updates.append('is_active = ?')
            params.append(1 if data.get('is_active') else 0)
        
        if not updates:
            return jsonify({'code': 0, 'msg': 'No fields to update', 'data': None}), 400
        
        # Add next_run_at update if interval was changed
        if next_run_interval is not None:
            updates.append(f"next_run_at = NOW() + INTERVAL '{next_run_interval} minutes'")
        
        updates.append('updated_at = NOW()')
        params.append(monitor_id)
        params.append(user_id)
        
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                f"UPDATE qd_position_monitors SET {', '.join(updates)} WHERE id = ? AND user_id = ?",
                params
            )
            db.commit()
            cur.close()

        return jsonify({'code': 1, 'msg': 'success', 'data': None})
    except Exception as e:
        logger.error(f"update_monitor failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


@portfolio_bp.route('/monitors/<int:monitor_id>', methods=['DELETE'])
@login_required
def delete_monitor(monitor_id):
    """Delete a monitor for the current user."""
    try:
        user_id = g.user_id
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                "DELETE FROM qd_position_monitors WHERE id = ? AND user_id = ?",
                (monitor_id, user_id)
            )
            db.commit()
            cur.close()

        return jsonify({'code': 1, 'msg': 'success', 'data': None})
    except Exception as e:
        logger.error(f"delete_monitor failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


@portfolio_bp.route('/monitors/<int:monitor_id>/run', methods=['POST'])
@login_required
def run_monitor_now(monitor_id):
    """Manually trigger a monitor to run immediately.
    
    Supports two modes:
    - async=true (default): Returns immediately, runs in background, notifies via notification system
    - async=false: Waits for completion and returns result (may timeout for large portfolios)
    """
    try:
        from app.services.portfolio_monitor import run_single_monitor
        
        user_id = g.user_id
        
        # Get parameters from request body
        data = request.get_json(force=True, silent=True) or {}
        language = data.get('language')
        async_mode = data.get('async', True)  # Default to async mode
        
        # Fallback to Accept-Language header for language
        if not language:
            accept_lang = request.headers.get('Accept-Language', '')
            if 'zh' in accept_lang.lower():
                language = 'zh-CN'
            else:
                language = 'en-US'
        
        if async_mode:
            # Async mode: Start background thread and return immediately
            import threading
            
            def run_in_background(mid, lang, uid):
                try:
                    run_single_monitor(mid, override_language=lang, user_id=uid)
                except Exception as e:
                    logger.error(f"Background monitor run failed: {e}")
            
            thread = threading.Thread(
                target=run_in_background,
                args=(monitor_id, language, user_id),
                daemon=True
            )
            thread.start()
            
            return jsonify({
                'code': 1, 
                'msg': 'success', 
                'data': {
                    'status': 'running',
                    'message': 'Monitor is running in background. Results will be sent via notification.'
                }
            })
        else:
            # Sync mode: Wait for completion (may timeout)
            result = run_single_monitor(monitor_id, override_language=language, user_id=user_id)
            return jsonify({'code': 1, 'msg': 'success', 'data': result})
            
    except Exception as e:
        logger.error(f"run_monitor_now failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


# ==================== Alerts CRUD ====================

@portfolio_bp.route('/alerts', methods=['GET'])
@login_required
def get_alerts():
    """Get all position alerts for the current user."""
    try:
        user_id = g.user_id
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                """
                SELECT a.id, a.position_id, a.market, a.symbol, a.alert_type, a.threshold,
                       a.notification_config, a.is_active, a.is_triggered, a.last_triggered_at,
                       a.trigger_count, a.repeat_interval, a.notes, a.created_at, a.updated_at,
                       p.name as position_name, p.side as position_side
                FROM qd_position_alerts a
                LEFT JOIN qd_manual_positions p ON a.position_id = p.id
                WHERE a.user_id = ?
                ORDER BY a.id DESC
                """,
                (user_id,)
            )
            rows = cur.fetchall() or []
            cur.close()

        alerts = []
        for row in rows:
            alerts.append({
                'id': row.get('id'),
                'position_id': row.get('position_id'),
                'market': row.get('market') or '',
                'symbol': row.get('symbol') or '',
                'alert_type': row.get('alert_type') or 'price_above',
                'threshold': float(row.get('threshold') or 0),
                'notification_config': _safe_json_loads(row.get('notification_config'), {}),
                'is_active': bool(row.get('is_active')),
                'is_triggered': bool(row.get('is_triggered')),
                'last_triggered_at': row.get('last_triggered_at'),
                'trigger_count': row.get('trigger_count') or 0,
                'repeat_interval': row.get('repeat_interval') or 0,
                'notes': row.get('notes') or '',
                'created_at': row.get('created_at'),
                'updated_at': row.get('updated_at'),
                'position_name': row.get('position_name') or '',
                'position_side': row.get('position_side') or 'long'
            })

        return jsonify({'code': 1, 'msg': 'success', 'data': alerts})
    except Exception as e:
        logger.error(f"get_alerts failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': []}), 500


@portfolio_bp.route('/alerts', methods=['POST'])
@login_required
def add_alert():
    """Add a new position alert for the current user."""
    try:
        user_id = g.user_id
        data = request.get_json() or {}
        position_id = data.get('position_id')  # Can be None for symbol-level alerts
        market = (data.get('market') or '').strip()
        symbol = _normalize_symbol(data.get('symbol'))
        alert_type = (data.get('alert_type') or 'price_above').strip()
        threshold = float(data.get('threshold') or 0)
        notification_config = data.get('notification_config') or {}
        is_active = bool(data.get('is_active', True))
        repeat_interval = int(data.get('repeat_interval') or 0)
        notes = (data.get('notes') or '').strip()
        
        # Validate alert_type
        valid_types = ('price_above', 'price_below', 'pnl_above', 'pnl_below')
        if alert_type not in valid_types:
            return jsonify({'code': 0, 'msg': f'Invalid alert_type. Must be one of: {valid_types}', 'data': None}), 400
        
        # If position_id provided, get market/symbol from position
        if position_id:
            with get_db_connection() as db:
                cur = db.cursor()
                cur.execute(
                    "SELECT market, symbol FROM qd_manual_positions WHERE id = ? AND user_id = ?",
                    (position_id, user_id)
                )
                pos = cur.fetchone()
                cur.close()
                if pos:
                    market = pos.get('market') or market
                    symbol = pos.get('symbol') or symbol
        
        if not market or not symbol:
            return jsonify({'code': 0, 'msg': 'Market and symbol are required', 'data': None}), 400
        
        if threshold <= 0 and alert_type.startswith('price_'):
            return jsonify({'code': 0, 'msg': 'Threshold must be positive for price alerts', 'data': None}), 400
        
        notification_config_json = json.dumps(notification_config if isinstance(notification_config, dict) else {}, ensure_ascii=False)
        
        with get_db_connection() as db:
            cur = db.cursor()
            
            # Check if alert already exists for this position (unique constraint)
            existing_alert_id = None
            if position_id:
                cur.execute(
                    "SELECT id FROM qd_position_alerts WHERE position_id = ? AND user_id = ?",
                    (position_id, user_id)
                )
                existing = cur.fetchone()
                if existing:
                    existing_alert_id = existing.get('id')
            
            if existing_alert_id:
                # Update existing alert instead of creating a new one
                cur.execute(
                    """
                    UPDATE qd_position_alerts 
                    SET alert_type = ?, threshold = ?, notification_config = ?, 
                        is_active = ?, is_triggered = 0, repeat_interval = ?, notes = ?, updated_at = NOW()
                    WHERE id = ?
                    """,
                    (alert_type, threshold, notification_config_json,
                     1 if is_active else 0, repeat_interval, notes, existing_alert_id)
                )
                alert_id = existing_alert_id
            else:
                # Create new alert
                cur.execute(
                    """
                    INSERT INTO qd_position_alerts 
                    (user_id, position_id, market, symbol, alert_type, threshold, notification_config, 
                     is_active, repeat_interval, notes, created_at, updated_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW(), NOW())
                    """,
                    (user_id, position_id, market, symbol, alert_type, threshold, notification_config_json,
                     1 if is_active else 0, repeat_interval, notes)
                )
                alert_id = cur.lastrowid
            
            db.commit()
            cur.close()

        return jsonify({'code': 1, 'msg': 'success', 'data': {'id': alert_id}})
    except Exception as e:
        logger.error(f"add_alert failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


@portfolio_bp.route('/alerts/<int:alert_id>', methods=['PUT'])
@login_required
def update_alert(alert_id):
    """Update an existing alert for the current user."""
    try:
        user_id = g.user_id
        data = request.get_json() or {}
        
        updates = []
        params = []
        
        if 'alert_type' in data:
            updates.append('alert_type = ?')
            params.append((data.get('alert_type') or 'price_above').strip())
        
        if 'threshold' in data:
            updates.append('threshold = ?')
            params.append(float(data.get('threshold') or 0))
        
        if 'notification_config' in data:
            notification_config = data.get('notification_config') or {}
            updates.append('notification_config = ?')
            params.append(json.dumps(notification_config if isinstance(notification_config, dict) else {}, ensure_ascii=False))
        
        if 'is_active' in data:
            updates.append('is_active = ?')
            params.append(1 if data.get('is_active') else 0)
            # Reset triggered state when re-activating
            if data.get('is_active'):
                updates.append('is_triggered = ?')
                params.append(0)
        
        if 'repeat_interval' in data:
            updates.append('repeat_interval = ?')
            params.append(int(data.get('repeat_interval') or 0))
        
        if 'notes' in data:
            updates.append('notes = ?')
            params.append((data.get('notes') or '').strip())
        
        if not updates:
            return jsonify({'code': 0, 'msg': 'No fields to update', 'data': None}), 400
        
        updates.append('updated_at = NOW()')
        params.append(alert_id)
        params.append(user_id)
        
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                f"UPDATE qd_position_alerts SET {', '.join(updates)} WHERE id = ? AND user_id = ?",
                params
            )
            db.commit()
            cur.close()

        return jsonify({'code': 1, 'msg': 'success', 'data': None})
    except Exception as e:
        logger.error(f"update_alert failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


@portfolio_bp.route('/alerts/<int:alert_id>', methods=['DELETE'])
@login_required
def delete_alert(alert_id):
    """Delete an alert for the current user."""
    try:
        user_id = g.user_id
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                "DELETE FROM qd_position_alerts WHERE id = ? AND user_id = ?",
                (alert_id, user_id)
            )
            db.commit()
            cur.close()

        return jsonify({'code': 1, 'msg': 'success', 'data': None})
    except Exception as e:
        logger.error(f"delete_alert failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


# ==================== Groups ====================

@portfolio_bp.route('/groups', methods=['GET'])
@login_required
def get_groups():
    """Get list of all groups with position counts for the current user."""
    try:
        user_id = g.user_id
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                """
                SELECT group_name, COUNT(*) as count
                FROM qd_manual_positions
                WHERE user_id = ? AND group_name != ''
                GROUP BY group_name
                ORDER BY group_name
                """,
                (user_id,)
            )
            rows = cur.fetchall() or []
            
            # Also get count of ungrouped
            cur.execute(
                "SELECT COUNT(*) as count FROM qd_manual_positions WHERE user_id = ? AND (group_name IS NULL OR group_name = '')",
                (user_id,)
            )
            ungrouped = cur.fetchone()
            cur.close()

        groups = []
        for row in rows:
            groups.append({
                'name': row.get('group_name'),
                'count': row.get('count') or 0
            })
        
        # Add ungrouped count
        ungrouped_count = (ungrouped.get('count') or 0) if ungrouped else 0
        
        return jsonify({
            'code': 1,
            'msg': 'success',
            'data': {
                'groups': groups,
                'ungrouped_count': ungrouped_count
            }
        })
    except Exception as e:
        logger.error(f"get_groups failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500


@portfolio_bp.route('/groups/rename', methods=['POST'])
@login_required
def rename_group():
    """Rename a group for the current user."""
    try:
        user_id = g.user_id
        data = request.get_json() or {}
        old_name = (data.get('old_name') or '').strip()
        new_name = (data.get('new_name') or '').strip()
        
        if not old_name:
            return jsonify({'code': 0, 'msg': 'old_name is required', 'data': None}), 400
        
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                "UPDATE qd_manual_positions SET group_name = ?, updated_at = NOW() WHERE user_id = ? AND group_name = ?",
                (new_name, user_id, old_name)
            )
            db.commit()
            cur.close()

        return jsonify({'code': 1, 'msg': 'success', 'data': None})
    except Exception as e:
        logger.error(f"rename_group failed: {str(e)}")
        logger.error(traceback.format_exc())
        return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500