feat: add Portfolio Management module with AI monitoring

backend_api_python/app/__init__.py

@@ -43,6 +43,30 @@ def get_reflection_worker():
     return _reflection_worker
 
 
+def start_portfolio_monitor():
+    """Start the portfolio monitor service if enabled.
+    
+    To enable it, set ENABLE_PORTFOLIO_MONITOR=true.
+    """
+    import os
+    enabled = os.getenv("ENABLE_PORTFOLIO_MONITOR", "true").lower() == "true"
+    if not enabled:
+        logger.info("Portfolio monitor is disabled. Set ENABLE_PORTFOLIO_MONITOR=true to enable.")
+        return
+    
+    # Avoid running twice with Flask reloader
+    debug = os.getenv("PYTHON_API_DEBUG", "false").lower() == "true"
+    if debug:
+        if os.environ.get("WERKZEUG_RUN_MAIN") != "true":
+            return
+    
+    try:
+        from app.services.portfolio_monitor import start_monitor_service
+        start_monitor_service()
+    except Exception as e:
+        logger.error(f"Failed to start portfolio monitor: {e}")
+
+
 def start_reflection_worker():
     """
     Start the reflection worker if enabled.
@@ -163,6 +187,7 @@ def create_app(config_name='default'):
     with app.app_context():
         start_pending_order_worker()
         start_reflection_worker()
+        start_portfolio_monitor()
         restore_running_strategies()
     
     return app

backend_api_python/app/routes/__init__.py

@@ -18,6 +18,7 @@ def register_routes(app: Flask):
     from app.routes.indicator import indicator_bp
     from app.routes.dashboard import dashboard_bp
     from app.routes.settings import settings_bp
+    from app.routes.portfolio import portfolio_bp
     
     app.register_blueprint(health_bp)
     app.register_blueprint(auth_bp, url_prefix='/api/user')  # 兼容前端 /api/user/login
@@ -31,4 +32,5 @@ def register_routes(app: Flask):
     app.register_blueprint(credentials_bp, url_prefix='/api/credentials')
     app.register_blueprint(dashboard_bp, url_prefix='/api/dashboard')
     app.register_blueprint(settings_bp, url_prefix='/api/settings')
+    app.register_blueprint(portfolio_bp, url_prefix='/api/portfolio')
 

backend_api_python/app/routes/portfolio.py

@@ -0,0 +1,991 @@
+"""
+Portfolio API routes (local-only).
+Manages manual positions (user's existing holdings) and AI monitoring tasks.
+"""
+from flask import Blueprint, request, jsonify
+import json
+import traceback
+import time
+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.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
+executor = ThreadPoolExecutor(max_workers=10)
+
+DEFAULT_USER_ID = 1
+
+
+def _now_ts() -> int:
+    return int(time.time())
+
+
+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) -> dict:
+    """Get price data for a single symbol."""
+    try:
+        cache_key = f"portfolio_price:{market}:{symbol}"
+        cached_data = cache.get(cache_key)
+        
+        if cached_data:
+            return {
+                'market': market,
+                'symbol': symbol,
+                'price': cached_data.get('price', 0),
+                'change': cached_data.get('change', 0),
+                'changePercent': cached_data.get('changePercent', 0)
+            }
+        
+        klines = kline_service.get_kline(market, symbol, '1D', 2)
+        
+        if klines and len(klines) > 0:
+            latest = klines[-1]
+            prev_close = klines[-2]['close'] if len(klines) > 1 else latest.get('open', 0)
+            current_price = latest.get('close', 0)
+            
+            change = round(current_price - prev_close, 4) if prev_close else 0
+            change_percent = round(change / prev_close * 100, 2) if prev_close and prev_close > 0 else 0
+            
+            result = {
+                'market': market,
+                'symbol': symbol,
+                'price': current_price,
+                'change': change,
+                'changePercent': change_percent
+            }
+            
+            cache.set(cache_key, result, 60)
+            return result
+        else:
+            return {
+                'market': market,
+                'symbol': symbol,
+                'price': 0,
+                'change': 0,
+                'changePercent': 0
+            }
+    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
+        }
+
+
+# ==================== Position CRUD ====================
+
+@portfolio_bp.route('/positions', methods=['GET'])
+def get_positions():
+    """Get all manual positions with current prices."""
+    try:
+        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
+                """,
+                (DEFAULT_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
+            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)
+                    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'])
+def add_position():
+    """Add a new manual position."""
+    try:
+        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)
+        now = _now_ts()
+        
+        with get_db_connection() as db:
+            cur = db.cursor()
+            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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(user_id, market, symbol, side) DO UPDATE SET
+                    name = excluded.name,
+                    quantity = excluded.quantity,
+                    entry_price = excluded.entry_price,
+                    entry_time = excluded.entry_time,
+                    notes = excluded.notes,
+                    tags = excluded.tags,
+                    group_name = excluded.group_name,
+                    updated_at = excluded.updated_at
+                """,
+                (DEFAULT_USER_ID, market, symbol, name, side, quantity, entry_price, entry_time, notes, tags_json, group_name, now, now)
+            )
+            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'])
+def update_position(position_id):
+    """Update an existing position."""
+    try:
+        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 = ?')
+        params.append(_now_ts())
+        params.append(position_id)
+        params.append(DEFAULT_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'])
+def delete_position(position_id):
+    """Delete a position."""
+    try:
+        with get_db_connection() as db:
+            cur = db.cursor()
+            cur.execute(
+                "DELETE FROM qd_manual_positions WHERE id = ? AND user_id = ?",
+                (position_id, DEFAULT_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'])
+def get_portfolio_summary():
+    """Get portfolio summary with total value, PnL, and market distribution."""
+    try:
+        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 = ?
+                """,
+                (DEFAULT_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
+        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)
+                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'])
+def get_monitors():
+    """Get all position monitors."""
+    try:
+        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
+                """,
+                (DEFAULT_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': row.get('last_run_at'),
+                'next_run_at': row.get('next_run_at'),
+                'last_result': _safe_json_loads(row.get('last_result'), {}),
+                'run_count': row.get('run_count') or 0,
+                'created_at': row.get('created_at'),
+                'updated_at': 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'])
+def add_monitor():
+    """Add a new position monitor."""
+    try:
+        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
+        now = _now_ts()
+        interval_minutes = int(config.get('interval_minutes') or 60)
+        next_run_at = now + (interval_minutes * 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (DEFAULT_USER_ID, name, position_ids_json, monitor_type, config_json, notification_config_json, 
+                 1 if is_active else 0, next_run_at, now, now)
+            )
+            monitor_id = cur.lastrowid
+            db.commit()
+            cur.close()
+
+        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'])
+def update_monitor(monitor_id):
+    """Update an existing monitor."""
+    try:
+        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())
+        
+        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
+            interval_minutes = int(config.get('interval_minutes') or 60)
+            updates.append('next_run_at = ?')
+            params.append(_now_ts() + (interval_minutes * 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
+        
+        updates.append('updated_at = ?')
+        params.append(_now_ts())
+        params.append(monitor_id)
+        params.append(DEFAULT_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'])
+def delete_monitor(monitor_id):
+    """Delete a monitor."""
+    try:
+        with get_db_connection() as db:
+            cur = db.cursor()
+            cur.execute(
+                "DELETE FROM qd_position_monitors WHERE id = ? AND user_id = ?",
+                (monitor_id, DEFAULT_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'])
+def run_monitor_now(monitor_id):
+    """Manually trigger a monitor to run immediately."""
+    try:
+        from app.services.portfolio_monitor import run_single_monitor
+        
+        # Get language from request body or Accept-Language header
+        data = request.get_json(force=True, silent=True) or {}
+        language = data.get('language')
+        
+        # Fallback to Accept-Language header
+        if not language:
+            accept_lang = request.headers.get('Accept-Language', '')
+            if 'zh' in accept_lang.lower():
+                language = 'en-US'
+            else:
+                language = 'en-US'
+        
+        result = run_single_monitor(monitor_id, override_language=language)
+        
+        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'])
+def get_alerts():
+    """Get all position alerts."""
+    try:
+        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
+                """,
+                (DEFAULT_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'])
+def add_alert():
+    """Add a new position alert."""
+    try:
+        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, DEFAULT_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
+        
+        now = _now_ts()
+        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, DEFAULT_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 = ?
+                    WHERE id = ?
+                    """,
+                    (alert_type, threshold, notification_config_json,
+                     1 if is_active else 0, repeat_interval, notes, now, 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    """,
+                    (DEFAULT_USER_ID, position_id, market, symbol, alert_type, threshold, notification_config_json,
+                     1 if is_active else 0, repeat_interval, notes, now, now)
+                )
+                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'])
+def update_alert(alert_id):
+    """Update an existing alert."""
+    try:
+        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 = ?')
+        params.append(_now_ts())
+        params.append(alert_id)
+        params.append(DEFAULT_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'])
+def delete_alert(alert_id):
+    """Delete an alert."""
+    try:
+        with get_db_connection() as db:
+            cur = db.cursor()
+            cur.execute(
+                "DELETE FROM qd_position_alerts WHERE id = ? AND user_id = ?",
+                (alert_id, DEFAULT_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'])
+def get_groups():
+    """Get list of all groups with position counts."""
+    try:
+        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
+                """,
+                (DEFAULT_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 = '')",
+                (DEFAULT_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'])
+def rename_group():
+    """Rename a group."""
+    try:
+        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
+        
+        now = _now_ts()
+        with get_db_connection() as db:
+            cur = db.cursor()
+            cur.execute(
+                "UPDATE qd_manual_positions SET group_name = ?, updated_at = ? WHERE user_id = ? AND group_name = ?",
+                (new_name, now, DEFAULT_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

backend_api_python/app/routes/strategy.py

@@ -752,4 +752,60 @@ def get_strategy_notifications():
     except Exception as e:
         logger.error(f"get_strategy_notifications failed: {str(e)}")
         logger.error(traceback.format_exc())
-        return jsonify({'code': 0, 'msg': str(e), 'data': {'items': []}}), 500
+        return jsonify({'code': 0, 'msg': str(e), 'data': {'items': []}}), 500
+
+
+@strategy_bp.route('/strategies/notifications/read', methods=['POST'])
+def mark_notification_read():
+    """Mark a single notification as read."""
+    try:
+        data = request.get_json(force=True, silent=True) or {}
+        notification_id = data.get('id')
+        if not notification_id:
+            return jsonify({'code': 0, 'msg': 'Missing id'}), 400
+
+        with get_db_connection() as db:
+            cur = db.cursor()
+            cur.execute(
+                "UPDATE qd_strategy_notifications SET is_read = 1 WHERE id = ?",
+                (int(notification_id),)
+            )
+            db.commit()
+            cur.close()
+
+        return jsonify({'code': 1, 'msg': 'success'})
+    except Exception as e:
+        logger.error(f"mark_notification_read failed: {str(e)}")
+        return jsonify({'code': 0, 'msg': str(e)}), 500
+
+
+@strategy_bp.route('/strategies/notifications/read-all', methods=['POST'])
+def mark_all_notifications_read():
+    """Mark all notifications as read."""
+    try:
+        with get_db_connection() as db:
+            cur = db.cursor()
+            cur.execute("UPDATE qd_strategy_notifications SET is_read = 1")
+            db.commit()
+            cur.close()
+
+        return jsonify({'code': 1, 'msg': 'success'})
+    except Exception as e:
+        logger.error(f"mark_all_notifications_read failed: {str(e)}")
+        return jsonify({'code': 0, 'msg': str(e)}), 500
+
+
+@strategy_bp.route('/strategies/notifications/clear', methods=['DELETE'])
+def clear_notifications():
+    """Clear all notifications (delete from database)."""
+    try:
+        with get_db_connection() as db:
+            cur = db.cursor()
+            cur.execute("DELETE FROM qd_strategy_notifications")
+            db.commit()
+            cur.close()
+
+        return jsonify({'code': 1, 'msg': 'success'})
+    except Exception as e:
+        logger.error(f"clear_notifications failed: {str(e)}")
+        return jsonify({'code': 0, 'msg': str(e)}), 500

backend_api_python/app/services/llm.py

@@ -37,7 +37,7 @@ class LLMService:
         config = load_addon_config()
         openrouter_config = config.get('openrouter', {})
         
-        default_model = openrouter_config.get('model', 'google/gemini-3-pro-preview')
+        default_model = openrouter_config.get('model', 'openai/gpt-4o')
         
         if model is None:
             model = default_model

backend_api_python/app/services/trading_executor.py

@@ -964,6 +964,17 @@ class TradingExecutor:
                             )
                             if ok:
                                 logger.info(f"Strategy {strategy_id} signal executed: {signal_type} @ {execute_price}")
+                                # Notify portfolio positions linked to this symbol
+                                try:
+                                    from app.services.portfolio_monitor import notify_strategy_signal_for_positions
+                                    notify_strategy_signal_for_positions(
+                                        market=market_type or 'Crypto',
+                                        symbol=symbol,
+                                        signal_type=signal_type,
+                                        signal_detail=f"策略: {strategy_name}\n信号: {signal_type}\n价格: {execute_price:.4f}"
+                                    )
+                                except Exception as link_e:
+                                    logger.warning(f"Strategy signal linkage notification failed: {link_e}")
                             else:
                                 logger.warning(f"Strategy {strategy_id} signal rejected/failed: {signal_type}")
 

backend_api_python/app/utils/db.py

@@ -242,6 +242,7 @@ def _init_db_schema(conn):
         "message": "TEXT DEFAULT ''",
         "payload_json": "TEXT DEFAULT ''",
         "created_at": "INTEGER",
+        "is_read": "INTEGER DEFAULT 0",
     })
 
     # 4. 指标代码表（参考 MySQL: qd_indicator_codes）
@@ -424,6 +425,119 @@ def _init_db_schema(conn):
         "updated_at": "INTEGER"
     })
 
+    # 11. Manual positions (user's existing holdings outside the system)
+    cursor.execute("""
+    CREATE TABLE IF NOT EXISTS qd_manual_positions (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        user_id INTEGER NOT NULL DEFAULT 1,
+        market TEXT NOT NULL,           -- Crypto/USStock/AShare/HShare/Forex/Futures
+        symbol TEXT NOT NULL,
+        name TEXT DEFAULT '',           -- Display name
+        side TEXT DEFAULT 'long',       -- long/short
+        quantity REAL NOT NULL DEFAULT 0,
+        entry_price REAL NOT NULL DEFAULT 0,
+        entry_time INTEGER,             -- Position open timestamp
+        notes TEXT DEFAULT '',          -- User notes
+        tags TEXT DEFAULT '',           -- JSON array of tags
+        group_name TEXT DEFAULT '',     -- Group name for categorization
+        created_at INTEGER,
+        updated_at INTEGER,
+        UNIQUE(user_id, market, symbol, side)
+    )
+    """)
+
+    ensure_columns("qd_manual_positions", {
+        "user_id": "INTEGER NOT NULL DEFAULT 1",
+        "market": "TEXT NOT NULL DEFAULT ''",
+        "symbol": "TEXT NOT NULL DEFAULT ''",
+        "name": "TEXT DEFAULT ''",
+        "side": "TEXT DEFAULT 'long'",
+        "quantity": "REAL NOT NULL DEFAULT 0",
+        "entry_price": "REAL NOT NULL DEFAULT 0",
+        "entry_time": "INTEGER",
+        "notes": "TEXT DEFAULT ''",
+        "tags": "TEXT DEFAULT ''",
+        "group_name": "TEXT DEFAULT ''",
+        "created_at": "INTEGER",
+        "updated_at": "INTEGER"
+    })
+    
+    # 11.1 Position alerts (price alerts and PnL alerts for manual positions)
+    cursor.execute("""
+    CREATE TABLE IF NOT EXISTS qd_position_alerts (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        user_id INTEGER NOT NULL DEFAULT 1,
+        position_id INTEGER,            -- FK to qd_manual_positions (NULL = symbol-level alert)
+        market TEXT DEFAULT '',
+        symbol TEXT DEFAULT '',
+        alert_type TEXT NOT NULL,       -- price_above / price_below / pnl_above / pnl_below
+        threshold REAL NOT NULL DEFAULT 0,
+        notification_config TEXT DEFAULT '',  -- JSON: channels, targets
+        is_active INTEGER DEFAULT 1,
+        is_triggered INTEGER DEFAULT 0,
+        last_triggered_at INTEGER,
+        trigger_count INTEGER DEFAULT 0,
+        repeat_interval INTEGER DEFAULT 0,    -- 0=once, >0=repeat every N seconds
+        notes TEXT DEFAULT '',
+        created_at INTEGER,
+        updated_at INTEGER
+    )
+    """)
+    
+    ensure_columns("qd_position_alerts", {
+        "user_id": "INTEGER NOT NULL DEFAULT 1",
+        "position_id": "INTEGER",
+        "market": "TEXT DEFAULT ''",
+        "symbol": "TEXT DEFAULT ''",
+        "alert_type": "TEXT NOT NULL DEFAULT 'price_above'",
+        "threshold": "REAL NOT NULL DEFAULT 0",
+        "notification_config": "TEXT DEFAULT ''",
+        "is_active": "INTEGER DEFAULT 1",
+        "is_triggered": "INTEGER DEFAULT 0",
+        "last_triggered_at": "INTEGER",
+        "trigger_count": "INTEGER DEFAULT 0",
+        "repeat_interval": "INTEGER DEFAULT 0",
+        "notes": "TEXT DEFAULT ''",
+        "created_at": "INTEGER",
+        "updated_at": "INTEGER"
+    })
+
+    # 12. Position monitors (AI analysis tasks for manual positions)
+    cursor.execute("""
+    CREATE TABLE IF NOT EXISTS qd_position_monitors (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        user_id INTEGER NOT NULL DEFAULT 1,
+        name TEXT DEFAULT '',                   -- Monitor name
+        position_ids TEXT DEFAULT '',           -- JSON array of position IDs (empty = all positions)
+        monitor_type TEXT DEFAULT 'ai',         -- ai / price_alert / pnl_alert
+        config TEXT DEFAULT '',                 -- JSON config (interval_minutes, prompt, thresholds, etc.)
+        notification_config TEXT DEFAULT '',    -- JSON notification config (channels, targets)
+        is_active INTEGER DEFAULT 1,
+        last_run_at INTEGER,
+        next_run_at INTEGER,
+        last_result TEXT DEFAULT '',            -- Last analysis result (JSON)
+        run_count INTEGER DEFAULT 0,
+        created_at INTEGER,
+        updated_at INTEGER
+    )
+    """)
+
+    ensure_columns("qd_position_monitors", {
+        "user_id": "INTEGER NOT NULL DEFAULT 1",
+        "name": "TEXT DEFAULT ''",
+        "position_ids": "TEXT DEFAULT ''",
+        "monitor_type": "TEXT DEFAULT 'ai'",
+        "config": "TEXT DEFAULT ''",
+        "notification_config": "TEXT DEFAULT ''",
+        "is_active": "INTEGER DEFAULT 1",
+        "last_run_at": "INTEGER",
+        "next_run_at": "INTEGER",
+        "last_result": "TEXT DEFAULT ''",
+        "run_count": "INTEGER DEFAULT 0",
+        "created_at": "INTEGER",
+        "updated_at": "INTEGER"
+    })
+
     conn.commit()
     logger.info("Database schema initialized (SQLite)")
 

backend_api_python/env.example

@@ -31,6 +31,14 @@ SQLITE_DATABASE_FILE=
 # Local mode default is enabled in code, but you can override here.
 ENABLE_PENDING_ORDER_WORKER=true
 
+# =========================
+# Portfolio monitor (optional)
+# =========================
+# Background service that runs scheduled AI analysis on manual positions
+# and sends notifications (email/telegram/browser).
+# Default: enabled. Set to false to disable.
+ENABLE_PORTFOLIO_MONITOR=true
+
 # Reclaim orders stuck in status=processing after worker crashes (seconds).
 PENDING_ORDER_STALE_SEC=90