diff --git a/README.md b/README.md index 2087e08..c70a8c7 100644 --- a/README.md +++ b/README.md @@ -6,9 +6,26 @@ ## docker部署教程2：https://www.bilibili.com/video/BV1j2FNz4EAi/ ## 股票知识讲解合集：https://www.bilibili.com/video/BV1Y2FGzzEeS/ ## 投资认知提升合集：https://www.bilibili.com/video/BV1ugBMBAEbW +## 价值投资核心逻辑：https://www.bilibili.com/video/BV1eJfxBrEjZ 如果你希望能在股市中长久生存下去，建议你能把上面的合集看完，会对你有很大帮助的！ +## ⭐ 2026.2.27更新 - 低估值价值投资策略 💎 + +**新增选股板块：基于价值投资核心逻辑的优选策略** + +基于视频[《头号投资法则》](https://www.bilibili.com/video/BV1eJfxBrEjZ)，通过低估值、高股息、低负债等多维度指标筛选安全边际极高的优质标的。 + +**核心功能：** +- 筛选条件：**低PE (≤20) + 低PB (≤1.5) + 高股息 (≥1%) + 低负债 (≤30%)** +- 排序机制：按流通市值从小到大排序，精准捕捉被错杀的小盘价值股 +- 量化择时： + - **买入**：每日扫描，开盘买入，单股限仓30%，最大持股4只。 + - **卖出**：持股满30天到期卖出，或 **RSI(14) > 70** 超买信号触发卖出。 +- 自动化工具：支持一键模拟买入、实时指标监测及 PDF/Markdown 报告导出。 + +--- + ## ⭐ 2026.2.27更新 - 宏观周期分析 🧭 **全新板块：康波周期 × 美林投资时钟 × 中国政策分析** diff --git a/app.py b/app.py index 4740011..a9af32f 100644 --- a/app.py +++ b/app.py @@ -288,7 +288,7 @@ def main(): if st.button("🏠 股票分析", width='stretch', key="nav_home", help="返回首页，进行单只股票的深度分析"): # 清除所有功能页面标志 for key in ['show_history', 'show_monitor', 'show_config', 'show_main_force', - 'show_sector_strategy', 'show_longhubang', 'show_portfolio', 'show_low_price_bull', 'show_news_flow', 'show_macro_cycle']: + 'show_sector_strategy', 'show_longhubang', 'show_portfolio', 'show_low_price_bull', 'show_news_flow', 'show_macro_cycle', 'show_value_stock']: if key in st.session_state: del st.session_state[key] @@ -322,7 +322,14 @@ def main(): if st.button("📈 净利增长", width='stretch', key="nav_profit_growth", help="净利润增长稳健股票筛选策略"): st.session_state.show_profit_growth = True for key in ['show_history', 'show_monitor', 'show_config', 'show_sector_strategy', - 'show_longhubang', 'show_portfolio', 'show_main_force', 'show_low_price_bull', 'show_small_cap', 'show_news_flow']: + 'show_longhubang', 'show_portfolio', 'show_main_force', 'show_low_price_bull', 'show_small_cap', 'show_news_flow', 'show_value_stock']: + if key in st.session_state: + del st.session_state[key] + + if st.button("💎 低估值策略", width='stretch', key="nav_value_stock", help="低PE+低PB+高股息+低负债 价值投资筛选"): + st.session_state.show_value_stock = True + for key in ['show_history', 'show_monitor', 'show_config', 'show_sector_strategy', + 'show_longhubang', 'show_portfolio', 'show_main_force', 'show_low_price_bull', 'show_small_cap', 'show_profit_growth', 'show_news_flow', 'show_macro_cycle']: if key in st.session_state: del st.session_state[key] @@ -519,6 +526,12 @@ def main(): display_profit_growth() return + # 检查是否显示低估值策略 + if 'show_value_stock' in st.session_state and st.session_state.show_value_stock: + from value_stock_ui import display_value_stock + display_value_stock() + return + # 检查是否显示智策板块 if 'show_sector_strategy' in st.session_state and st.session_state.show_sector_strategy: display_sector_strategy() diff --git a/sector_strategy.db b/sector_strategy.db index c092157..dc39e4f 100644 Binary files a/sector_strategy.db and b/sector_strategy.db differ diff --git a/value_stock_selector.py b/value_stock_selector.py new file mode 100644 index 0000000..ab492e8 --- /dev/null +++ b/value_stock_selector.py @@ -0,0 +1,175 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +低估值选股模块 +使用pywencai获取低估值优质股票 +""" + +import pandas as pd +import pywencai +from datetime import datetime +from typing import Tuple, Optional +import time + + +class ValueStockSelector: + """低估值选股类""" + + def __init__(self): + self.raw_data = None + self.selected_stocks = None + + def get_value_stocks(self, top_n: int = 10) -> Tuple[bool, Optional[pd.DataFrame], str]: + """ + 获取低估值优质股票 + + 选股策略： + - 市盈率 ≤ 20 + - 市净率 ≤ 1.5 + - 股息率 ≥ 1% + - 资产负债率 ≤ 30% + - 非ST + - 非科创板 + - 非创业板 + - 按流通市值由小到大排名 + + Args: + top_n: 返回前N只股票 + + Returns: + (success, dataframe, message) + """ + try: + print(f"\n{'='*60}") + print(f"💎 低估值选股 - 数据获取中") + print(f"{'='*60}") + print(f"策略: PE≤20 + PB≤1.5 + 股息率≥1% + 资产负债率≤30%") + print(f"排除: ST、科创板、创业板") + print(f"排序: 按流通市值由小到大") + print(f"目标: 筛选前{top_n}只股票") + + # 构建问财查询语句 + query = ( + "市盈率小于等于20，" + "市净率小于等于1.5，" + "股息率大于等于1%，" + "资产负债率小于等于30%，" + "非st，" + "非科创板，" + "非创业板，" + "按流通市值由小到大排名" + ) + + print(f"\n查询语句: {query}") + print(f"正在调用问财接口...") + + # 调用pywencai + result = pywencai.get(query=query, loop=True) + + if result is None: + return False, None, "问财接口返回None，请检查网络或稍后重试" + + # 转换为DataFrame + df_result = self._convert_to_dataframe(result) + + if df_result is None or df_result.empty: + return False, None, "未获取到符合条件的股票数据" + + print(f"✅ 成功获取 {len(df_result)} 只股票") + + # 显示获取到的列名 + print(f"\n获取到的数据字段:") + for col in df_result.columns[:15]: + print(f" - {col}") + if len(df_result.columns) > 15: + print(f" ... 还有 {len(df_result.columns) - 15} 个字段") + + # 保存原始数据 + self.raw_data = df_result + + # 取前N只 + if len(df_result) > top_n: + selected = df_result.head(top_n) + print(f"\n从 {len(df_result)} 只股票中选出前 {top_n} 只") + else: + selected = df_result + print(f"\n共 {len(df_result)} 只符合条件的股票") + + self.selected_stocks = selected + + # 显示选中的股票 + print(f"\n✅ 选中的股票:") + for idx, row in selected.iterrows(): + code = row.get('股票代码', 'N/A') + name = row.get('股票简称', 'N/A') + pe = row.get('市盈率', row.get('市盈率(动态)', 'N/A')) + pb = row.get('市净率', 'N/A') + div_rate = row.get('股息率', 'N/A') + debt_ratio = row.get('资产负债率', 'N/A') + cap = row.get('流通市值', 'N/A') + print(f" {idx+1}. {code} {name} - PE:{pe} PB:{pb} 股息率:{div_rate}% 负债率:{debt_ratio}% 流通市值:{cap}") + + print(f"{'='*60}\n") + + return True, selected, f"成功筛选出{len(selected)}只低估值优质股票" + + except Exception as e: + error_msg = f"获取数据失败: {str(e)}" + print(f"❌ {error_msg}") + import traceback + traceback.print_exc() + return False, None, error_msg + + def _convert_to_dataframe(self, result) -> Optional[pd.DataFrame]: + """将pywencai返回结果转换为DataFrame""" + try: + if isinstance(result, pd.DataFrame): + return result + elif isinstance(result, dict): + if 'data' in result: + return pd.DataFrame(result['data']) + elif 'result' in result: + return pd.DataFrame(result['result']) + else: + return pd.DataFrame(result) + elif isinstance(result, list): + return pd.DataFrame(result) + else: + print(f"⚠️ 未知的数据格式: {type(result)}") + return None + except Exception as e: + print(f"转换DataFrame失败: {e}") + return None + + def get_stock_codes(self) -> list: + """ + 获取选中股票的代码列表（去掉市场后缀） + + Returns: + 股票代码列表 + """ + if self.selected_stocks is None or self.selected_stocks.empty: + return [] + + codes = [] + for code in self.selected_stocks['股票代码'].tolist(): + if isinstance(code, str): + clean_code = code.split('.')[0] if '.' in code else code + codes.append(clean_code) + else: + codes.append(str(code)) + + return codes + + +# 测试 +if __name__ == "__main__": + print("=" * 60) + print("测试低估值选股模块") + print("=" * 60) + + selector = ValueStockSelector() + success, df, msg = selector.get_value_stocks(top_n=10) + print(f"\n结果: {msg}") + if success and df is not None: + print(f"共 {len(df)} 只股票") diff --git a/value_stock_strategy.py b/value_stock_strategy.py new file mode 100644 index 0000000..1b8c4cb --- /dev/null +++ b/value_stock_strategy.py @@ -0,0 +1,291 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +低估值量化交易策略 +实现基于持股周期和RSI超买的买卖择时策略 +""" + +import pandas as pd +import akshare as ak +from datetime import datetime, timedelta +from typing import Dict, List, Optional +import logging + + +class ValueStockStrategy: + """低估值量化交易策略""" + + 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.3 # 个股最大仓位30% + self.max_daily_buy = 2 # 单日最大买入数 + self.holding_period = 30 # 持股周期（天） + self.rsi_period = 14 # RSI计算周期 + self.rsi_overbought = 70 # RSI超买阈值 + + # 持仓信息 + 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: + """ + 检查是否可以买入 + + 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: + """ + 计算买入数量 + + 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: + """ + 执行买入操作 + + Returns: + (是否成功, 消息, 交易详情) + """ + can, reason = self.can_buy(stock_code) + if not can: + return False, reason, None + + shares, amount = self.calculate_buy_amount(price) + if shares == 0: + return False, "资金不足以买入1手", None + + # 更新持仓 + self.positions[stock_code] = { + 'name': stock_name, + 'buy_price': price, + 'shares': shares, + 'amount': amount, + 'buy_date': date, + 'holding_days': 0 + } + + self.available_cash -= amount + self.daily_buy_count += 1 + + trade = { + 'action': '买入', + 'code': stock_code, + 'name': stock_name, + 'price': price, + 'shares': shares, + 'amount': amount, + 'date': date, + 'reason': '开盘买入信号' + } + self.trade_history.append(trade) + + msg = f"买入 {stock_code} {stock_name} {shares}股 @ {price}元, 金额: {amount:.2f}元" + return True, msg, trade + + def calculate_rsi(self, stock_code: str) -> Optional[float]: + """ + 计算股票的RSI指标 + + Args: + stock_code: 股票代码 + + Returns: + RSI值 或 None + """ + try: + # 获取近60天日线数据 + df = ak.stock_zh_a_hist( + symbol=stock_code, + period="daily", + start_date=(datetime.now() - timedelta(days=90)).strftime("%Y%m%d"), + end_date=datetime.now().strftime("%Y%m%d"), + adjust="qfq" + ) + + if df is None or len(df) < self.rsi_period + 1: + return None + + # 计算RSI + close = df['收盘'].astype(float) + delta = close.diff() + gain = delta.where(delta > 0, 0) + loss = (-delta).where(delta < 0, 0) + + avg_gain = gain.rolling(window=self.rsi_period).mean() + avg_loss = loss.rolling(window=self.rsi_period).mean() + + rs = avg_gain / avg_loss + rsi = 100 - (100 / (1 + rs)) + + latest_rsi = rsi.iloc[-1] + return round(float(latest_rsi), 2) if pd.notna(latest_rsi) else None + + except Exception as e: + self.logger.warning(f"RSI计算失败 {stock_code}: {e}") + return None + + def should_sell(self, stock_code: str, current_date: str = None) -> tuple: + """ + 判断是否应该卖出 + + 策略： + 1. 持股满30天强制卖出 + 2. RSI超买（>70）卖出 + + Returns: + (是否卖出, 原因, RSI值) + """ + if stock_code not in self.positions: + return False, "未持有该股票", None + + position = self.positions[stock_code] + position['holding_days'] += 1 + + # 条件1：持股满30天 + if position['holding_days'] >= self.holding_period: + return True, f"持股满{self.holding_period}天，到期卖出", None + + # 条件2：RSI超买 + rsi = self.calculate_rsi(stock_code) + if rsi is not None and rsi > self.rsi_overbought: + return True, f"RSI={rsi} 超买（>{self.rsi_overbought}），卖出离场", rsi + + return False, f"继续持有 (已持{position['holding_days']}天, RSI={rsi})", rsi + + def sell(self, stock_code: str, price: float, date: str, reason: str = "") -> tuple: + """ + 执行卖出操作 + + Returns: + (是否成功, 消息, 交易详情) + """ + if stock_code not in self.positions: + return False, "未持有该股票", None + + position = self.positions[stock_code] + amount = position['shares'] * price + profit = amount - position['amount'] + profit_pct = (price - position['buy_price']) / position['buy_price'] * 100 + + trade = { + 'action': '卖出', + 'code': stock_code, + 'name': position['name'], + 'price': price, + 'shares': position['shares'], + 'amount': amount, + 'date': date, + 'buy_price': position['buy_price'], + 'profit': profit, + 'profit_pct': round(profit_pct, 2), + 'holding_days': position['holding_days'], + 'reason': reason + } + self.trade_history.append(trade) + + self.available_cash += amount + del self.positions[stock_code] + + emoji = "🟢" if profit >= 0 else "🔴" + msg = f"{emoji} 卖出 {stock_code} {position['name']} {position['shares']}股 @ {price}元, 盈亏: {profit:.2f}元 ({profit_pct:+.2f}%), 原因: {reason}" + return True, msg, trade + + def get_portfolio_summary(self) -> Dict: + """获取投资组合摘要""" + total_position_value = sum( + pos['shares'] * pos['buy_price'] for pos in self.positions.values() + ) + total_assets = self.available_cash + total_position_value + + # 统计交易 + sells = [t for t in self.trade_history if t['action'] == '卖出'] + total_profit = sum(t.get('profit', 0) for t in sells) + win_trades = sum(1 for t in sells if t.get('profit', 0) > 0) + total_trades = len(sells) + win_rate = (win_trades / total_trades * 100) if total_trades > 0 else 0 + + return { + 'initial_capital': self.initial_capital, + 'available_cash': round(self.available_cash, 2), + 'position_value': round(total_position_value, 2), + 'total_assets': round(total_assets, 2), + 'total_return': round((total_assets - self.initial_capital) / self.initial_capital * 100, 2), + 'total_profit': round(total_profit, 2), + 'holding_count': len(self.positions), + 'max_stocks': self.max_stocks, + 'total_trades': total_trades, + 'win_trades': win_trades, + 'win_rate': round(win_rate, 2) + } + + def get_positions(self) -> List[Dict]: + """获取当前持仓列表""" + positions = [] + for code, pos in self.positions.items(): + positions.append({ + 'code': code, + 'name': pos['name'], + 'buy_price': pos['buy_price'], + 'shares': pos['shares'], + 'amount': pos['amount'], + 'buy_date': pos['buy_date'], + 'holding_days': pos['holding_days'] + }) + return positions + + def get_trade_history(self) -> List[Dict]: + """获取交易历史""" + return self.trade_history diff --git a/value_stock_ui.py b/value_stock_ui.py new file mode 100644 index 0000000..e99a137 --- /dev/null +++ b/value_stock_ui.py @@ -0,0 +1,420 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +低估值策略UI模块 +""" + +import streamlit as st +import pandas as pd +from datetime import datetime +from value_stock_selector import ValueStockSelector +from value_stock_strategy import ValueStockStrategy + + +def display_value_stock(): + """显示低估值选股界面""" + + st.markdown(""" +
+

+ 💎 低估值策略 - 价值投资选股 +

+

+ 基于视频 头号投资法则 +

+

+ 低PE + 低PB + 高股息 + 低负债 — 寻找被市场低估的优质标的 +

+
+ """, unsafe_allow_html=True) + + st.markdown("---") + + st.markdown(""" + ### 📋 选股策略说明 + + **筛选条件**： + - ✅ 市盈率（PE）≤ 20 + - ✅ 市净率（PB）≤ 1.5 + - ✅ 股息率 ≥ 1% + - ✅ 资产负债率 ≤ 30% + - ✅ 非ST股票 + - ✅ 非科创板 + - ✅ 非创业板 + - ✅ 按流通市值由小到大排名 + + **量化交易策略**： + - 💰 资金量：100万元 + - 📈 买入时机：开盘买入 + - 💼 单股最大仓位：30% + - 🎯 最大持股数：4只 + - 🛒 每日最多买入：2只 + - 📉 卖出条件①：持股满30天到期卖出 + - 📉 卖出条件②：RSI超买（>70）卖出 + """) + + st.markdown("---") + + # 参数设置 + col1, col2 = st.columns([2, 1]) + + with col1: + top_n = st.slider( + "筛选数量", + min_value=5, + max_value=20, + value=10, + step=1, + help="选择展示的股票数量", + key="value_stock_top_n" + ) + + with col2: + st.info(f"💡 将筛选流通市值最小的前{top_n}只低估值股票") + + st.markdown("---") + + # 开始选股按钮 + if st.button("🚀 开始低估值选股", type="primary", width='content', key="value_stock_start"): + + with st.spinner("正在获取数据，请稍候..."): + selector = ValueStockSelector() + success, stocks_df, message = selector.get_value_stocks(top_n=top_n) + + if success and stocks_df is not None: + st.session_state.value_stocks = stocks_df + st.session_state.value_stock_selector = selector + st.success(f"✅ {message}") + st.rerun() + else: + st.error(f"❌ {message}") + + # 显示选股结果 + if 'value_stocks' in st.session_state: + display_stock_results( + st.session_state.value_stocks, + st.session_state.get('value_stock_selector') + ) + + +def display_stock_results(stocks_df: pd.DataFrame, selector): + """显示选股结果""" + + st.markdown("---") + st.markdown("## 📊 选股结果") + + # 统计信息 + col1, col2, col3, col4 = st.columns(4) + + with col1: + st.metric("筛选数量", f"{len(stocks_df)} 只") + + with col2: + pe_col = None + for pattern in ['市盈率', '市盈率(动态)']: + matching = [col for col in stocks_df.columns if pattern in col] + if matching: + pe_col = matching[0] + break + if pe_col: + valid = pd.to_numeric(stocks_df[pe_col], errors='coerce').dropna() + if len(valid) > 0: + st.metric("平均PE", f"{valid.mean():.1f}") + else: + st.metric("平均PE", "-") + else: + st.metric("平均PE", "-") + + with col3: + pb_col = None + matching = [col for col in stocks_df.columns if '市净率' in col] + if matching: + pb_col = matching[0] + valid = pd.to_numeric(stocks_df[pb_col], errors='coerce').dropna() + if len(valid) > 0: + st.metric("平均PB", f"{valid.mean():.2f}") + else: + st.metric("平均PB", "-") + else: + st.metric("平均PB", "-") + + with col4: + div_col = None + matching = [col for col in stocks_df.columns if '股息率' in col] + if matching: + div_col = matching[0] + valid = pd.to_numeric(stocks_df[div_col], errors='coerce').dropna() + if len(valid) > 0: + st.metric("平均股息率", f"{valid.mean():.2f}%") + else: + st.metric("平均股息率", "-") + else: + st.metric("平均股息率", "-") + + st.markdown("---") + + # 显示股票列表 + st.markdown("### 📋 精选低估值股票") + + for idx, row in stocks_df.iterrows(): + code = row.get('股票代码', 'N/A') + name = row.get('股票简称', 'N/A') + + # 获取关键指标用于标题 + pe_val = '' + for pattern in ['市盈率', '市盈率(动态)']: + matching = [col for col in stocks_df.columns if pattern in col] + if matching: + v = row.get(matching[0]) + if v is not None and not pd.isna(v): + try: + pe_val = f" PE:{float(v):.1f}" + except: + pass + break + + pb_val = '' + matching = [col for col in stocks_df.columns if '市净率' in col] + if matching: + v = row.get(matching[0]) + if v is not None and not pd.isna(v): + try: + pb_val = f" PB:{float(v):.2f}" + except: + pass + + with st.expander( + f"【第{idx+1}名】{code} - {name}{pe_val}{pb_val}", + expanded=(idx < 3) + ): + display_stock_detail(row, stocks_df) + + # 完整数据表格 + st.markdown("---") + st.markdown("### 📊 完整数据表格") + + # 选择关键列 + display_cols = ['股票代码', '股票简称'] + for pattern in ['最新价', '股价']: + matching = [col for col in stocks_df.columns if pattern in col] + if matching: + display_cols.append(matching[0]) + break + for pattern in ['市盈率', '市净率', '股息率', '资产负债率', '流通市值', '所属行业']: + matching = [col for col in stocks_df.columns if pattern in col] + if matching: + display_cols.append(matching[0]) + + final_cols = [col for col in display_cols if col in stocks_df.columns] + + if final_cols: + st.dataframe(stocks_df[final_cols], width='content', height=400) + + csv = stocks_df[final_cols].to_csv(index=False, encoding='utf-8-sig') + st.download_button( + label="📥 下载股票列表CSV", + data=csv, + file_name=f"value_stock_{datetime.now().strftime('%Y%m%d')}.csv", + mime="text/csv", + key="value_csv_download" + ) + + # 量化交易模拟 + st.markdown("---") + display_strategy_simulation(stocks_df, selector) + + +def display_stock_detail(row: pd.Series, df: pd.DataFrame): + """显示单个股票详情""" + + def is_valid(value): + if value is None: + return False + if isinstance(value, float) and pd.isna(value): + return False + if isinstance(value, str) and value.strip() in ('', 'N/A', 'nan', 'None'): + return False + return True + + def fmt(value, suffix=''): + if not is_valid(value): + return "-" + try: + return f"{float(value):.2f}{suffix}" + except: + return str(value) + suffix + + # 基本估值数据 + col1, col2, col3, col4 = st.columns(4) + + with col1: + for p in ['市盈率', '市盈率(动态)']: + m = [c for c in df.columns if p in c] + if m: + st.metric("📊 市盈率(PE)", fmt(row.get(m[0]))) + break + + with col2: + m = [c for c in df.columns if '市净率' in c] + if m: + st.metric("📊 市净率(PB)", fmt(row.get(m[0]))) + + with col3: + m = [c for c in df.columns if '股息率' in c] + if m: + st.metric("💰 股息率", fmt(row.get(m[0]), '%')) + + with col4: + m = [c for c in df.columns if '资产负债率' in c] + if m: + st.metric("📉 资产负债率", fmt(row.get(m[0]), '%')) + + # 补充信息 + st.markdown("**其他指标**：") + info_parts = [] + for pattern in ['最新价', '股价', '流通市值', '总市值', '所属行业', '涨跌幅']: + m = [c for c in df.columns if pattern in c] + if m: + val = row.get(m[0]) + if is_valid(val): + info_parts.append(f"**{pattern}**: {val}") + if info_parts: + st.markdown(" | ".join(info_parts)) + + +def display_strategy_simulation(stocks_df: pd.DataFrame, selector): + """显示量化交易策略模拟""" + + st.markdown("## 🎯 策略模拟") + + st.info(""" + **策略规则**： + - 📈 **买入**：开盘价买入，单股最大仓位30%，每日最多买2只 + - 📉 **卖出条件①**：持股满30天，到期自动卖出 + - 📉 **卖出条件②**：RSI(14) > 70 超买，触发卖出 + - 🎯 **最大持股**：4只 + - 💰 **初始资金**：100万元 + """) + + col1, col2 = st.columns(2) + + with col1: + if st.button("🎮 开始策略模拟", type="primary", width='content', key="value_sim_start"): + st.session_state.show_value_strategy_sim = True + + with col2: + pass + + if st.session_state.get('show_value_strategy_sim'): + run_strategy_simulation(stocks_df) + + +def run_strategy_simulation(stocks_df: pd.DataFrame): + """运行策略模拟""" + + st.markdown("---") + st.markdown("### 📈 策略模拟执行") + + strategy = ValueStockStrategy(initial_capital=1000000.0) + + # 模拟买入 + st.markdown("#### 1️⃣ 模拟买入信号") + + buy_results = [] + current_date = datetime.now().strftime("%Y-%m-%d") + + for idx, row in stocks_df.head(strategy.max_daily_buy).iterrows(): + code = str(row.get('股票代码', '')).split('.')[0] + name = row.get('股票简称', 'N/A') + + # 尝试获取价格 + price = 0 + for p in ['最新价', '股价']: + m = [c for c in stocks_df.columns if p in c] + if m: + try: + price = float(row.get(m[0], 0)) + except: + pass + if price > 0: + break + + if price > 0: + success, message, trade = strategy.buy(code, name, price, current_date) + buy_results.append({ + 'success': success, + 'message': message, + 'trade': trade + }) + + for result in buy_results: + if result['success']: + st.success(result['message']) + else: + st.warning(f"⚠️ {result['message']}") + + # RSI检查 + st.markdown("---") + st.markdown("#### 2️⃣ RSI卖出信号检测") + + with st.spinner("正在计算RSI指标..."): + for code, pos in list(strategy.positions.items()): + rsi = strategy.calculate_rsi(code) + if rsi is not None: + if rsi > strategy.rsi_overbought: + st.warning(f"⚠️ {code} {pos['name']} RSI={rsi} > {strategy.rsi_overbought}，触发超买卖出信号！") + else: + st.info(f"ℹ️ {code} {pos['name']} RSI={rsi}，正常范围") + else: + st.info(f"ℹ️ {code} {pos['name']} RSI计算中...") + + # 显示持仓 + st.markdown("---") + st.markdown("#### 3️⃣ 当前持仓") + + positions = strategy.get_positions() + if positions: + positions_df = pd.DataFrame(positions) + st.dataframe(positions_df, width='content') + else: + st.info("暂无持仓") + + # 显示账户摘要 + st.markdown("---") + st.markdown("#### 4️⃣ 账户摘要") + + summary = strategy.get_portfolio_summary() + + col1, col2, col3, col4 = st.columns(4) + + with col1: + st.metric("初始资金", f"{summary['initial_capital']:,.0f} 元") + with col2: + st.metric("可用资金", f"{summary['available_cash']:,.0f} 元") + with col3: + st.metric("持仓市值", f"{summary['position_value']:,.0f} 元") + with col4: + st.metric("总资产", f"{summary['total_assets']:,.0f} 元") + + st.markdown("---") + st.markdown("#### 📝 策略说明") + st.markdown(""" + **后续操作**： + 1. **持有期管理**：系统跟踪每只股票的持有天数（30天到期） + 2. **RSI监测**：每日收盘后计算RSI(14) + - RSI > 70：超买信号，提示卖出 + - RSI < 30：超卖信号（可作为加仓参考） + 3. **轮动买入**：卖出后释放资金，继续买入新的低估值股票 + + **风险提示**： + - ⚠️ 本策略为模拟演示，实际交易存在滑点、手续费等成本 + - ⚠️ 低估值不代表没有风险，价值陷阱需警惕 + - ⚠️ 请谨慎评估风险，理性投资 + """) + + +# 主入口 +if __name__ == "__main__": + display_value_stock()