aiagents-stock/low_price_bull_strategy.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
低价擒牛量化交易策略
实现基于MA均线的买卖择时策略
"""

import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import logging


class LowPriceBullStrategy:
    """低价擒牛量化交易策略"""
    
    def __init__(self, initial_capital: float = 1000000.0):
        """
        初始化策略
        
        Args:
            initial_capital: 初始资金（默认100万）
        """
        self.logger = logging.getLogger(__name__)
        
        # 策略参数
        self.initial_capital = initial_capital
        self.available_cash = initial_capital
        self.max_stocks = 4  # 账户最大持股数
        self.max_position_per_stock = 0.4  # 个股最大持仓比例（4成）
        self.max_daily_buy = 2  # 单日最大买入数
        self.holding_period = 5  # 持股周期（天）
        
        # 持仓信息
        self.positions: Dict[str, Dict] = {}  # {股票代码: {买入价, 数量, 买入日期, 持有天数}}
        self.trade_history: List[Dict] = []  # 交易历史
        
        # 当日交易计数
        self.daily_buy_count = 0
        self.current_date = None
    
    def reset_daily_counter(self, date):
        """重置当日计数器"""
        if self.current_date != date:
            self.current_date = date
            self.daily_buy_count = 0
    
    def can_buy(self, stock_code: str) -> tuple[bool, str]:
        """
        检查是否可以买入
        
        Returns:
            (是否可买, 原因)
        """
        # 检查是否已持有
        if stock_code in self.positions:
            return False, "已持有该股票"
        
        # 检查持股数量
        if len(self.positions) >= self.max_stocks:
            return False, f"已达最大持股数限制({self.max_stocks}只)"
        
        # 检查当日买入数量
        if self.daily_buy_count >= self.max_daily_buy:
            return False, f"今日已达最大买入数限制({self.max_daily_buy}只)"
        
        # 检查资金
        if self.available_cash <= 0:
            return False, "可用资金不足"
        
        return True, "可以买入"
    
    def calculate_buy_amount(self, stock_price: float) -> tuple[int, float]:
        """
        计算买入数量（满仓策略）
        
        Args:
            stock_price: 股票价格
            
        Returns:
            (买入股数, 买入金额)
        """
        # 满仓策略：使用所有可用资金
        max_amount = self.available_cash
        
        # 但不能超过个股最大持仓
        max_per_stock = self.initial_capital * self.max_position_per_stock
        target_amount = min(max_amount, max_per_stock)
        
        # 计算股数（A股100股为1手）
        shares = int(target_amount / stock_price / 100) * 100
        
        if shares < 100:
            return 0, 0
        
        actual_amount = shares * stock_price
        return shares, actual_amount
    
    def buy(self, stock_code: str, stock_name: str, price: float, date: str) -> tuple[bool, str, Optional[Dict]]:
        """
        执行买入操作
        
        Returns:
            (是否成功, 消息, 交易详情)
        """
        # 重置当日计数
        self.reset_daily_counter(date)
        
        # 检查是否可买入
        can_buy, reason = self.can_buy(stock_code)
        if not can_buy:
            return False, reason, None
        
        # 计算买入数量
        shares, amount = self.calculate_buy_amount(price)
        
        if shares == 0:
            return False, "资金不足，无法买入100股", None
        
        # 执行买入
        self.positions[stock_code] = {
            'name': stock_name,
            'shares': shares,
            'buy_price': price,
            'buy_date': date,
            'holding_days': 0
        }
        
        self.available_cash -= amount
        self.daily_buy_count += 1
        
        # 记录交易
        trade = {
            'type': 'BUY',
            'code': stock_code,
            'name': stock_name,
            'price': price,
            'shares': shares,
            'amount': amount,
            'date': date,
            'cash_after': self.available_cash
        }
        self.trade_history.append(trade)
        
        message = f"✅ 买入成功: {stock_code} {stock_name} | 价格:{price:.2f} | 数量:{shares}股 | 金额:{amount:.2f}元"
        self.logger.info(message)
        
        return True, message, trade
    
    def should_sell(self, stock_code: str, ma5: float, ma20: float, current_date: str) -> tuple[bool, str]:
        """
        判断是否应该卖出
        
        策略：
        1. MA5下穿MA20时卖出
        2. 持股满5天强制卖出
        
        Returns:
            (是否卖出, 原因)
        """
        if stock_code not in self.positions:
            return False, "未持有该股票"
        
        position = self.positions[stock_code]
        
        # 更新持有天数
        # 简化处理，按交易日计算
        position['holding_days'] += 1
        
        # 检查持股周期
        if position['holding_days'] >= self.holding_period:
            return True, f"持股满{self.holding_period}天，到期卖出"
        
        # 检查MA5下穿MA20
        if ma5 is not None and ma20 is not None:
            if ma5 < ma20:
                return True, "MA5下穿MA20，技术信号卖出"
        
        return False, "持有"
    
    def sell(self, stock_code: str, price: float, date: str, reason: str = "") -> tuple[bool, str, Optional[Dict]]:
        """
        执行卖出操作
        
        Returns:
            (是否成功, 消息, 交易详情)
        """
        if stock_code not in self.positions:
            return False, "未持有该股票", None
        
        position = self.positions[stock_code]
        shares = position['shares']
        buy_price = position['buy_price']
        
        # 计算盈亏
        amount = shares * price
        cost = shares * buy_price
        profit = amount - cost
        profit_pct = (profit / cost) * 100 if cost > 0 else 0
        
        # 归还资金
        self.available_cash += amount
        
        # 移除持仓
        del self.positions[stock_code]
        
        # 记录交易
        trade = {
            'type': 'SELL',
            'code': stock_code,
            'name': position['name'],
            'price': price,
            'shares': shares,
            'amount': amount,
            'date': date,
            'reason': reason,
            'buy_price': buy_price,
            'profit': profit,
            'profit_pct': profit_pct,
            'cash_after': self.available_cash
        }
        self.trade_history.append(trade)
        
        profit_str = f"+{profit:.2f}" if profit >= 0 else f"{profit:.2f}"
        message = f"✅ 卖出成功: {stock_code} {position['name']} | 价格:{price:.2f} | 数量:{shares}股 | 盈亏:{profit_str}元({profit_pct:+.2f}%) | 原因:{reason}"
        self.logger.info(message)
        
        return True, message, trade
    
    def get_portfolio_summary(self) -> Dict:
        """
        获取投资组合摘要
        
        Returns:
            组合摘要信息
        """
        # 计算持仓市值（需要当前价格，这里用买入价估算）
        position_value = sum(
            pos['shares'] * pos['buy_price'] 
            for pos in self.positions.values()
        )
        
        total_value = self.available_cash + position_value
        
        # 计算收益
        total_profit = total_value - self.initial_capital
        total_profit_pct = (total_profit / self.initial_capital) * 100
        
        return {
            'initial_capital': self.initial_capital,
            'available_cash': self.available_cash,
            'position_value': position_value,
            'total_value': total_value,
            'total_profit': total_profit,
            'total_profit_pct': total_profit_pct,
            'positions_count': len(self.positions),
            'max_stocks': self.max_stocks,
            'trade_count': len(self.trade_history)
        }
    
    def get_positions(self) -> List[Dict]:
        """获取当前持仓列表"""
        return [
            {
                'code': code,
                'name': pos['name'],
                'shares': pos['shares'],
                'buy_price': pos['buy_price'],
                'buy_date': pos['buy_date'],
                'holding_days': pos['holding_days']
            }
            for code, pos in self.positions.items()
        ]
    
    def get_trade_history(self) -> List[Dict]:
        """获取交易历史"""
        return self.trade_history.copy()