""" 新星1995 - 股票模拟交易游戏 基于 Tushare 数据的股票模拟交易系统 """ import subprocess import sys import time import os # ===== 0. 环境准备：自动安装缺失的库 ===== REQUIRED = [("tushare", "tushare"), ("pandas", "pandas"), ("numpy", "numpy"), ("tabulate", "tabulate")] for pkg, imp in REQUIRED: try: __import__(imp) except ImportError: print(f"正在安装 {pkg} ...") subprocess.check_call([sys.executable, "-m", "pip", "install", pkg, "-q"]) import tushare as ts import pandas as pd import numpy as np from tabulate import tabulate import random from datetime import datetime, timedelta # ===== Tushare 初始化 ===== TOKEN = "d4ade4dc2337bb63733c3ec90deb49bfe013ac6b81452bdba73641e7" ts.set_token(TOKEN) pro = ts.pro_api() # ===== 工具函数 ===== def cls(): os.system("cls" if os.name == "nt" else "clear") def input_default(prompt, default): """带默认值的输入""" val = input(f" {prompt} [默认: {default}]: ").strip() if not val: return default try: return type(default)(val) except Exception: print(f" 输入无效，使用默认值: {default}") return default def api(func, **kw): """带限速的 tushare API 调用""" time.sleep(0.35) try: return func(**kw) except Exception as e: print(f"\n [API 错误] {e}") return pd.DataFrame() def find_nearest_trade_date(df, target, direction="after"): """在 trade_date 列中找到最近的交易日""" df = df.sort_values("trade_date") if direction == "after": after = df[df["trade_date"] >= target] return after.iloc[0]["trade_date"] if not after.empty else None else: before = df[df["trade_date"] <= target] return before.iloc[-1]["trade_date"] if not before.empty else None # ════════════════════════════════════════════════════════════ # 一、设定筛选板块 # ════════════════════════════════════════════════════════════ def get_settings(): cls() print("=" * 60) print(" ★ 新 星 1 9 9 5 ★") print(" 股票模拟交易游戏") print("=" * 60) print("\n【设定筛选条件】（直接回车使用默认值）\n") s = {} s["listing_year"] = input_default("X 年前上市", 2015) s["max_pb"] = input_default("历史最大 PB 上限", 7.2) s["min_bps"] = input_default("历史最小每股净资产下限", 2.0) s["min_beta"] = input_default("贝塔系数下限", 0.5) s["min_roe"] = input_default("历史最小 ROE 下限 (%)", -50.0) s["pb_amp"] = input_default("快进 PB 振幅 (%)", 5.0) s["sim_date"] = input_default("模拟时间点 (YYYYMMDD)", "20220401") s["pb_amp_dec"] = s["pb_amp"] / 100.0 # 小数形式 return s # ════════════════════════════════════════════════════════════ # 筛选 & 选股 # ════════════════════════════════════════════════════════════ def fetch_index_cache(settings): """缓存上证指数的日线数据（beta 计算 + 展示用）""" cache = {} # beta 用：模拟时间点前 400 天 end_dt = datetime.strptime(settings["sim_date"], "%Y%m%d") start_beta = (end_dt - timedelta(days=420)).strftime("%Y%m%d") idx = api(pro.index_daily, ts_code="000001.SH", start_date=start_beta, end_date=settings["sim_date"]) if not idx.empty: idx = idx.sort_values("trade_date").reset_index(drop=True) cache["beta_period"] = idx[["trade_date", "close", "pct_chg"]].copy() else: cache["beta_period"] = pd.DataFrame() return cache def check_criteria(ts_code, settings, idx_cache): """逐条检查某只股票是否满足全部筛选条件，满足则返回附加数据""" sim_date = settings["sim_date"] # ---- 1. PB 历史 ---- pb_all = api(pro.daily_basic, ts_code=ts_code, start_date="19900101", end_date=sim_date) if pb_all.empty or "pb" not in pb_all.columns: return None pb_all = pb_all.dropna(subset=["pb"]) pb_all = pb_all[pb_all["pb"] > 0].copy() if pb_all.empty: return None if pb_all["pb"].max() >= settings["max_pb"]: return None # ---- 2. 财务指标（ROE、每股净资产）---- fina = api(pro.fina_indicator, ts_code=ts_code) if fina.empty or "end_date" not in fina.columns: return None fina_y = fina[fina["end_date"].str.endswith("1231")].copy() fina_y = fina_y.drop_duplicates(subset=["end_date"], keep="first") if fina_y.empty: return None # 每股净资产 bps_vals = fina_y.dropna(subset=["bps"]) if bps_vals.empty or bps_vals["bps"].min() < settings["min_bps"]: return None # ROE（tushare 已是百分比形式，如 15.5 = 15.5%） roe_vals = fina_y.dropna(subset=["roe"]) if roe_vals.empty or roe_vals["roe"].min() < settings["min_roe"]: return None # ---- 3. 贝塔系数 ---- end_dt = datetime.strptime(sim_date, "%Y%m%d") start_dt = (end_dt - timedelta(days=400)).strftime("%Y%m%d") stock_d = api(pro.daily, ts_code=ts_code, start_date=start_dt, end_date=sim_date) if stock_d.empty or "pct_chg" not in stock_d.columns: return None idx_beta = idx_cache.get("beta_period", pd.DataFrame()) if idx_beta.empty: return None merged = pd.merge( stock_d[["trade_date", "pct_chg"]], idx_beta[["trade_date", "pct_chg"]], on="trade_date", suffixes=("_s", "_i"), ).dropna() if len(merged) < 30: return None cov = np.cov(merged["pct_chg_s"].values, merged["pct_chg_i"].values) beta = cov[0][1] / cov[1][1] if cov[1][1] != 0 else 0 if beta < settings["min_beta"]: return None # 全部条件满足 pb_all = pb_all.sort_values("trade_date").reset_index(drop=True) fina_y = fina_y.sort_values("end_date").reset_index(drop=True) return {"beta": round(beta, 4), "pb_hist": pb_all, "fina_annual": fina_y} def filter_and_select(settings, idx_cache): """筛选并随机选取一只符合条件的股票""" print("\n正在获取股票列表 ...") stocks = api(pro.stock_basic, exchange="", list_status="L", fields="ts_code,symbol,name,industry,list_date") if stocks.empty: return None cutoff = f"{settings['listing_year']}0101" stocks = stocks[stocks["list_date"] < cutoff].reset_index(drop=True) print(f" {settings['listing_year']} 年前上市: {len(stocks)} 只\n") if stocks.empty: return None candidates = stocks.sample(frac=1).reset_index(drop=True) max_try = min(80, len(candidates)) for i in range(max_try): row = candidates.iloc[i] tc, nm = row["ts_code"], row["name"] print(f"\r [{i+1}/{max_try}] 正在检查 {nm}({tc}) ", end="", flush=True) try: extra = check_criteria(tc, settings, idx_cache) if extra is not None: print(f"\n\n ✓ 符合条件: {nm} ({tc})\n") return {**row.to_dict(), **extra} except Exception: continue print(f"\n\n ✗ 检查了 {max_try} 只股票，未找到全部符合条件的股票。") return None # ════════════════════════════════════════════════════════════ # 二、信息展示板块 # ════════════════════════════════════════════════════════════ def display_info(stock, settings): cls() sim_date = settings["sim_date"] pb_hist = stock["pb_hist"] fina = stock["fina_annual"] print("=" * 60) print(" 【股票信息展示】") print("=" * 60) # 模拟时间点 PB td = find_nearest_trade_date(pb_hist, sim_date, "after") if td is None: td = find_nearest_trade_date(pb_hist, sim_date, "before") sim_pb = pb_hist.loc[pb_hist["trade_date"] == td, "pb"].values sim_pb = round(sim_pb[0], 4) if len(sim_pb) else "N/A" # 上证指数 idx = api(pro.index_daily, ts_code="000001.SH", start_date=sim_date, end_date=sim_date) if idx.empty: _s = (datetime.strptime(sim_date, "%Y%m%d") - timedelta(days=10)).strftime("%Y%m%d") idx = api(pro.index_daily, ts_code="000001.SH", start_date=_s, end_date=sim_date) if not idx.empty: idx = idx.sort_values("trade_date") sh_close = round(idx.iloc[-1]["close"], 2) if not idx.empty else "N/A" info_tbl = [ ["股票名称", stock["name"]], ["股票代码", stock["ts_code"]], ["所属行业", stock["industry"]], ["贝塔系数", stock["beta"]], ["模拟时间点 PB", sim_pb], ["模拟时间点上证指数", sh_close], ] print("\n" + tabulate(info_tbl, tablefmt="grid")) # 年度 PB / ROE sim_year = int(sim_date[:4]) years = list(range(2015, sim_year)) rows = [] for y in years: yp = pb_hist[(pb_hist["trade_date"] >= f"{y}0101") & (pb_hist["trade_date"] <= f"{y}1231")] min_pb = round(yp["pb"].min(), 4) if not yp.empty else "-" avg_pb = round(yp["pb"].mean(), 4) if not yp.empty else "-" max_pb = round(yp["pb"].max(), 4) if not yp.empty else "-" yf = fina[fina["end_date"] == f"{y}1231"] roe_v = round(yf["roe"].values[0], 2) if not yf.empty and pd.notna(yf["roe"].values[0]) else "-" roe_s = f"{roe_v}%" if roe_v != "-" else "-" rows.append([y, min_pb, avg_pb, max_pb, roe_s]) print(f"\n 2015 ~ {sim_year - 1} 年度数据:\n") print(tabulate(rows, headers=["年份", "最小PB", "平均PB", "最大PB", "ROE"], tablefmt="grid")) print(f"\n [Y] 进入模拟买卖 | [N] 重新筛选 | [其他] 退出") return input(" 请输入: ").strip().upper() # ════════════════════════════════════════════════════════════ # 三、模拟买卖板块 # ════════════════════════════════════════════════════════════ def compute_trigger_points(pb_df, amplitude): """根据 PB 振幅计算所有触发点""" if pb_df.empty: return [] points = [] last_pb = None for _, r in pb_df.iterrows(): cur_pb = r["pb"] if last_pb is None: points.append(r) last_pb = cur_pb elif last_pb > 0 and abs(cur_pb - last_pb) / last_pb >= amplitude: points.append(r) last_pb = cur_pb return points def run_trading(stock, settings, game_state): cls() print("=" * 60) print(" 【模拟买卖】") print("=" * 60) print(f" 股票: {stock['name']} ({stock['ts_code']})") print(f" 快进 PB 振幅: {settings['pb_amp']}%\n") print(" 正在加载交易数据 ...") sim_date = settings["sim_date"] ts_code = stock["ts_code"] end_date = datetime.now().strftime("%Y%m%d") # PB 数据（模拟时间点开始） pb_future = api(pro.daily_basic, ts_code=ts_code, start_date=sim_date, end_date=end_date) if pb_future.empty or "pb" not in pb_future.columns: print(" 无法获取交易数据！") input(" 按回车返回 ...") return None pb_future = pb_future.dropna(subset=["pb"]) pb_future = pb_future[pb_future["pb"] > 0].copy() pb_future = pb_future.sort_values("trade_date").reset_index(drop=True) if pb_future.empty: print(" 该时间段无有效 PB 数据！") input(" 按回车返回 ...") return None # 上证指数 idx_f = api(pro.index_daily, ts_code="000001.SH", start_date=sim_date, end_date=end_date) idx_map = {} if not idx_f.empty: idx_map = dict(zip(idx_f["trade_date"], idx_f["close"])) # 计算触发点 triggers = compute_trigger_points(pb_future, settings["pb_amp_dec"]) if not triggers: print(" 没有触发点数据！") input(" 按回车返回 ...") return None print(f" 已计算 {len(triggers)} 个模拟买卖点\n") # 交易状态 position = 0.0 # 当前持仓 (0~100) avg_cost = 0.0 # 加权平均成本 total_weighted = 0.0 # position * avg_cost 的累计 cur_idx = 0 traded = False sells = [] # (数量, PB, 当时avg_cost) buys = [] # (数量, PB) first_date = triggers[0]["trade_date"] while True: pt = triggers[cur_idx] t_date = pt["trade_date"] t_pb = pt["pb"] sh_val = idx_map.get(t_date, "-") if sh_val != "-": sh_val = round(sh_val, 2) # 交易天数 days_count = len(pb_future[pb_future["trade_date"] <= t_date]) # 当前收益率 ret_str = "-" if avg_cost > 0: ret_str = f"{t_pb / avg_cost:.4f}" # 显示 tbl = [ ["模拟买卖点日期", t_date], ["模拟买卖点 PB", round(t_pb, 4)], ["上证指数", sh_val], ["持有仓位", f"{position:.1f}%"], ["平均成本", f"{avg_cost:.4f}" if avg_cost > 0 else "-"], ["当前交易天数", days_count], ["当前收益率", ret_str], ] print(" " + "─" * 46) print(tabulate(tbl, tablefmt="grid")) print(f"\n [K] 下一买卖点 [B/数量] 买入 [S/数量] 卖出 [N] 结束(需清仓)") cmd = input(" 指令> ").strip().upper() # ---- K: 下一触发点 ---- if cmd == "K": if cur_idx < len(triggers) - 1: cur_idx += 1 else: print(" ⚠ 已到达最后一个交易触发点！") # ---- B/数量: 买入 ---- elif cmd.startswith("B/") or cmd.startswith("B\\"): try: amt = float(cmd[2:]) if amt <= 0: print(" ⚠ 数量必须 > 0"); continue if position + amt > 100: print(f" ⚠ 仓位不能超100%！当前{position:.1f}%，最多买入{100-position:.1f}%"); continue total_weighted += amt * t_pb position += amt avg_cost = total_weighted / position buys.append((amt, t_pb)) traded = True print(f" ✓ 买入 {amt}% PB={t_pb:.4f} 仓位={position:.1f}% 平均成本={avg_cost:.4f}") except ValueError: print(" ⚠ 格式错误！示例: B/40") # ---- S/数量: 卖出 ---- elif cmd.startswith("S/") or cmd.startswith("S\\"): try: amt = float(cmd[2:]) if amt <= 0: print(" ⚠ 数量必须 > 0"); continue if position <= 0: print(" ⚠ 当前无持仓！"); continue actual = min(amt, position) sells.append((actual, t_pb, avg_cost)) position -= actual if position <= 0.001: position = 0.0 total_weighted = 0.0 avg_cost = 0.0 else: total_weighted = avg_cost * position traded = True print(f" ✓ 卖出 {actual}% PB={t_pb:.4f} 剩余仓位={position:.1f}%") if amt > actual: print(f" （超出持仓，已卖出全部 {actual}%）") except ValueError: print(" ⚠ 格式错误！示例: S/40") # ---- N: 结束 ---- elif cmd == "N": if position > 0.001: print(f" ⚠ 当前仓位 {position:.1f}%，必须清仓后才能结束！") else: last_date = t_date break else: print(" ⚠ 无效指令") # 计算本轮结果 if traded: game_state["total_stocks"] += 1 # 本轮收益率 = Σ(卖出仓位% × (卖出PB / 卖出时avg_cost − 1)) round_return = 0.0 for s_amt, s_pb, s_avg in sells: if s_avg > 0: round_return += s_amt * (s_pb / s_avg - 1) # round_return 单位是"百分比点" # 本轮交易天数 start_i = pb_future[pb_future["trade_date"] >= first_date].index end_i = pb_future[pb_future["trade_date"] <= last_date].index round_days = (end_i[-1] - start_i[0] + 1) if len(start_i) and len(end_i) else 0 return { "round_return": round_return, "round_days": int(round_days), "traded": traded, } # ════════════════════════════════════════════════════════════ # 四、结算信息板块 # ════════════════════════════════════════════════════════════ def show_settlement(result, game_state): cls() print("=" * 60) print(" 【结算信息】") print("=" * 60) rr = result["round_return"] rd = result["round_days"] # 累积收益率（乘算） game_state["cum_return"] *= (1 + rr / 100) game_state["cum_days"] += rd cum_pct = (game_state["cum_return"] - 1) * 100 tbl = [ ["累积交易股票数量", game_state["total_stocks"]], ["本轮收益率", f"{rr:.2f}%"], ["本轮交易天数", rd], ["累积收益率", f"{cum_pct:.2f}%"], ["累积交易天数", game_state["cum_days"]], ] print("\n" + tabulate(tbl, tablefmt="grid")) print(f"\n [Y] 记录信息，重新筛选 | [N] 退出游戏") return input(" 请输入: ").strip().upper() # ════════════════════════════════════════════════════════════ # 主程序 # ════════════════════════════════════════════════════════════ def main(): settings = get_settings() game_state = { "cum_return": 1.0, "cum_days": 0, "total_stocks": 0, } print("\n 正在初始化上证指数数据 ...") idx_cache = fetch_index_cache(settings) while True: stock = filter_and_select(settings, idx_cache) if stock is None: print("\n 无法找到符合条件的股票，请调整筛选条件后重试。") break choice = display_info(stock, settings) if choice == "Y": result = run_trading(stock, settings, game_state) if result is None: continue choice = show_settlement(result, game_state) if choice != "Y": print("\n 游戏结束，感谢游玩！") break elif choice == "N": continue else: print("\n 游戏结束，感谢游玩！") break if __name__ == "__main__": main()