重要提示：本指标公式源码属于技术指标，仅作为知识分享和学习参考，不做为投资依据。
炒股最累的是什么？不是买卖，是从5000多只股票里选出那几只真强势的。 你看，单按“创新高”筛一遍，呼啦啦出来几百只，根本看不过来。为啥？因为很多创新高只是个股自己的事，跟大盘、跟板块没半毛钱关系，持续性极差。 但如果一只股票创新高，它所在的板块也正在创新高，那性质就完全变了——这说明是整个板块在走强，资金在集体做多，个股不是孤立无援。 先找强势板块，再找强势股。 板块和个股一起创新高，才叫真强势 别人给你一堆名单，它给你一条逻辑。这个工具干的就是这件事： 扫全市场400多个板块，先揪出正在创新高的板块，再看这些板块里哪些个股也在同步创新高。 板块和个股共振，比你在5000只里瞎找靠谱一百倍。 你不用全市场大海捞针，只在最强板块里挑。选股范围一下从5000缩到几十只，效率直接拉满。
# -*- coding: utf-8 -*-"""板块创新高扫描器 —— 桌面 GUI 版（不依赖网页）==============================================基于 tkinter（Python 自带，无需联网打开浏览器），界面包含： - 2 个判定选项：用【收盘价】创高 / 用【最高价】创高 - 手动输入：板块创 N 日新高 - 手动输入：个股创 M 日新高 ← 板块与个股周期独立 - 每板块取前 N 只、限制创新高板块数 - 开始扫描按钮 + 进度条，结果直接在窗口表格展示（板块可展开看个股）复用 sector_new_high.py 的扫描内核（行情/板块解析/新高判定）。运行： python sector_new_high_gui.py # 双击即可 python sector_new_high_gui.py --tdx-path "D:/tdx""""from __future__ import annotationsimport argparseimport osimport queueimport sysimport threadingimport timefrom datetime import datetime# ---- 让双击（系统 python）也能找到 venv 里的 eltdx -------------------------_HERE = os.path.dirname(os.path.abspath(__file__))_VENV_SP = os.path.join(_HERE, ".venv_eltdx", "Lib", "site-packages")if os.path.isdir(_VENV_SP) and _VENV_SP not in sys.path: sys.path.insert(0, _VENV_SP)import tkinter as tkimport tkinter.font as tkfontfrom tkinter import ttk, messagebox, filedialogimport sector_new_high as core # 复用扫描内核# ---------------------------------------------------------------------------# 全局字体（统一放大）# ---------------------------------------------------------------------------FONT_TITLE = ("Microsoft YaHei", 18, "bold")FONT_SUBTITLE = ("Microsoft YaHei", 14, "bold")FONT_LABEL = ("Microsoft YaHei", 12)FONT_RADIO = ("Microsoft YaHei", 12)FONT_BTN = ("Microsoft YaHei", 12)FONT_STATUS = ("Microsoft YaHei", 11)FONT_TREE = ("Microsoft YaHei", 12)FONT_TREE_HEAD = ("Microsoft YaHei", 12, "bold")# ---------------------------------------------------------------------------# 扫描工作线程（通过 queue 与主线程通信，避免卡 UI）# ---------------------------------------------------------------------------class ScanWorker(threading.Thread): def __init__(self, q: queue.Queue, params: dict): super().__init__(daemon=True) self.q = q self.p = params def run(self): try: self._run() except Exception as e: # 任何异常都回传，避免无声失败 self.q.put(("error", f"{type(e).__name__}: {e}")) def _run(self): q = self.q p = self.p # 先做一次协议级主站探测：剔除「TCP 能连但协议不通」的失效节点， # 否则会卡在失效主机上反复失败。结果进程内缓存，只探测一次。 q.put(("status", "正在探测可用行情主站...")) hosts = core.resolve_hosts() q.put(("status", f"可用行情主站 {len(hosts)} 个，正在查找板块文件...")) block_path = core.find_block_file(p.get("tdx_path")) if not block_path: self.q.put(("error", "未找到板块文件 infoharbor_block.dat，请用「通达信目录」按钮指定。")) return q.put(("status", f"板块文件: {block_path}")) blocks = core.parse_blocks(block_path) q.put(("status", f"解析到 {len(blocks)} 个板块，加载代码表...")) core.build_name_map() q.put(("status", f"名称映射 {len(core._NAME_MAP)} 条，开始扫描板块...")) sectors = core.load_sector_indices() sectors = [(c6, fc, nm) for (c6, fc, nm) in sectors if c6 in blocks] total = len(sectors) q.put(("phase", "sector", total, 0)) # 阶段：板块，总数 lookback_sec = p["lookback_sector"] lookback_stk = p["lookback_stock"] use_close = p["use_close"] count_sec = lookback_sec + 6 count_stk = lookback_stk + 6 # ---- 第一层：板块创新高（板块用 lookback_sector） ---- new_secs = [] done = 0 lock = threading.Lock() def eval_sector(item): c6, fc, nm = item s = core.fetch_kline(fc, count_sec, "index") if s is None: return None is_new, prev_max, bp, last = core.new_high_info(s, lookback_sec, use_close) if not is_new: return None return {"code": c6, "full_code": fc, "name": nm, "breakout_pct": bp, "last": last, "prev_max": prev_max} with core.cf.ThreadPoolExecutor(max_workers=core.THREADS) as ex: import concurrent.futures as cf_local futs = {ex.submit(eval_sector, it): it for it in sectors} for fut in cf_local.as_completed(futs): r = fut.result() with lock: done += 1 if r: new_secs.append(r) q.put(("prog", done, total, len(new_secs))) new_secs.sort(key=lambda x: x["breakout_pct"], reverse=True) max_sec = p["max_sectors"] if max_sec and len(new_secs) > max_sec: new_secs = new_secs[:max_sec] q.put(("status", f"创新高板块 {len(new_secs)} 个，开始扫描成分股...")) # ---- 第二层：成分股创新高（个股用 lookback_stock） ---- sum_members = sum(min(len(blocks[s["code"]]["members"]), p["member_cap"] or 10**9) for s in new_secs) q.put(("phase", "stock", sum_members, 0)) report = [] m_done = 0 m_lock = threading.Lock() def eval_stock(full): s = core.fetch_kline(full, count_stk, "stock") if s is None: return None is_new, prev_max, bp, last = core.new_high_info(s, lookback_stk, use_close) if not is_new: return None return {"full_code": full, "name": core._NAME_MAP.get(full, ""), "last": last, "breakout_pct": bp, "prev_max": prev_max} for sec in new_secs: members = blocks[sec["code"]]["members"] if p["member_cap"] and len(members) > p["member_cap"]: members = members[:p["member_cap"]] stocks = [] sec_done = 0 with core.cf.ThreadPoolExecutor(max_workers=core.THREADS) as ex: futs = {ex.submit(eval_stock, m): m for m in members} for fut in cf_local.as_completed(futs): r = fut.result() with m_lock: m_done += 1 sec_done += 1 if r: stocks.append(r) q.put(("prog2", m_done, sum_members, sec["name"], sec_done, len(members))) stocks.sort(key=lambda x: x["breakout_pct"], reverse=True) sec["stocks"] = stocks[:p["top_n"]] sec["stock_count"] = len(stocks) sec["member_total"] = len(members) report.append(sec) q.put(("done", report, time.time()))# ---------------------------------------------------------------------------# GUI# ---------------------------------------------------------------------------class App: def __init__(self, root: tk.Tk, tdx_path: str | None): self.root = root self.tdx_path = tdx_path self.scanning = False self.q = queue.Queue() self.worker = None root.title("板块创新高扫描器") root.geometry("1020x720") try: from ctypes import windll windll.shcore.SetProcessDpiAwareness(1) except Exception: pass self._setup_styles() self._build_widgets() self._poll() # ---- ttk 样式统一放大 ---- def _setup_styles(self): style = ttk.Style() style.configure("TLabel", font=FONT_LABEL) style.configure("TButton", font=FONT_BTN) style.configure("TRadiobutton", font=FONT_RADIO) style.configure("TSpinbox", font=FONT_LABEL) style.configure("TEntry", font=FONT_LABEL) style.configure("TLabelframe.Label", font=FONT_LABEL) style.configure("Treeview", font=FONT_TREE, rowheight=28) style.configure("Treeview.Heading", font=FONT_TREE_HEAD) # ---- 界面布局 ---- def _build_widgets(self): pad = {"padx": 12, "pady": 6} # 顶部说明 ttk.Label(self.root, text="板块创新高扫描 · 桌面版（无需浏览器）", font=FONT_TITLE).pack(anchor="w", **pad) # 参数区 frm = ttk.LabelFrame(self.root, text="扫描参数") frm.pack(fill="x", **pad) # 2 个判定选项（收盘价 / 最高价） opt = ttk.LabelFrame(frm, text="新高判定口径（2 选 1）") opt.grid(row=0, column=0, columnspan=3, sticky="ew", padx=8, pady=8) self.use_close = tk.BooleanVar(value=True) ttk.Radiobutton(opt, text="收盘价创高", variable=self.use_close, value=True).pack(side="left", padx=14, pady=6) ttk.Radiobutton(opt, text="最高价创高", variable=self.use_close, value=False).pack(side="left", padx=14, pady=6) # 板块创 N 日新高（手动输入） ttk.Label(frm, text="板块创多少日新高：").grid(row=1, column=0, sticky="w", padx=8, pady=8) self.sec_days = tk.IntVar(value=250) self.sec_spin = ttk.Spinbox(frm, from_=5, to=999, width=8, textvariable=self.sec_days, font=FONT_LABEL) self.sec_spin.grid(row=1, column=1, sticky="w", padx=8, pady=8) ttk.Label(frm, text="日").grid(row=1, column=2, sticky="w", padx=2) # 个股创 M 日新高（手动输入） ttk.Label(frm, text="个股创多少日新高：").grid(row=2, column=0, sticky="w", padx=8, pady=8) self.stk_days = tk.IntVar(value=250) self.stk_spin = ttk.Spinbox(frm, from_=5, to=999, width=8, textvariable=self.stk_days, font=FONT_LABEL) self.stk_spin.grid(row=2, column=1, sticky="w", padx=8, pady=8) ttk.Label(frm, text="日").grid(row=2, column=2, sticky="w", padx=2) # 每板块取前 N 只 + 限制板块数 + 通达信目录 ttk.Label(frm, text="每板块取前 N 只：").grid(row=3, column=0, sticky="w", padx=8, pady=8) self.top_n = tk.IntVar(value=5) ttk.Spinbox(frm, from_=1, to=30, width=8, textvariable=self.top_n, font=FONT_LABEL).grid(row=3, column=1, sticky="w", padx=8, pady=8) ttk.Label(frm, text="限制创新高板块数：").grid(row=4, column=0, sticky="w", padx=8, pady=8) self.max_sec = tk.IntVar(value=0) ttk.Spinbox(frm, from_=0, to=200, width=8, textvariable=self.max_sec, font=FONT_LABEL).grid(row=4, column=1, sticky="w", padx=8, pady=8) ttk.Label(frm, text="0 = 不限").grid(row=4, column=2, sticky="w", padx=2) ttk.Label(frm, text="通达信目录：").grid(row=5, column=0, sticky="w", padx=8, pady=8) self.tdx_var = tk.StringVar(value=self.tdx_path or "") ttk.Entry(frm, textvariable=self.tdx_var, width=40, font=FONT_LABEL).grid( row=5, column=1, sticky="w", padx=8, pady=8) ttk.Button(frm, text="浏览...", command=self._browse).grid(row=5, column=2, sticky="w", padx=4) # 控制按钮 ctrl = ttk.Frame(self.root) ctrl.pack(fill="x", **pad) self.start_btn = ttk.Button(ctrl, text="开始扫描", command=self._start) self.start_btn.pack(side="left", padx=(0, 10)) self.stop_btn = ttk.Button(ctrl, text="停止", command=self._stop, state="disabled") self.stop_btn.pack(side="left") # 进度条 + 状态 self.pb = ttk.Progressbar(self.root, mode="determinate", maximum=100, value=0) self.pb.pack(fill="x", **pad) self.status = ttk.Label(self.root, text="就绪。点击「开始扫描」。", foreground="#555", font=FONT_STATUS) self.status.pack(anchor="w", **pad) # 结果树 cols = ("name", "code", "price", "break", "ratio") self.tree = ttk.Treeview(self.root, columns=cols, show="tree headings", height=16) self.tree.heading("#0", text="板块 / 个股") self.tree.column("#0", width=220, minwidth=140) self.tree.heading("name", text="名称") self.tree.column("name", width=160) self.tree.heading("code", text="代码") self.tree.column("code", width=110) self.tree.heading("price", text="现价") self.tree.column("price", width=90) self.tree.heading("break", text="突破幅度") self.tree.column("break", width=100) self.tree.heading("ratio", text="成分股新高") self.tree.column("ratio", width=130) self.tree.tag_configure("sector", background="#eef2f7", font=("Microsoft YaHei", 13, "bold")) self.tree.tag_configure("stock", foreground="#222", font=FONT_TREE) self.tree.pack(fill="both", expand=True, **pad) # 滚动条 vsb = ttk.Scrollbar(self.tree, orient="vertical", command=self.tree.yview) self.tree.configure(yscrollcommand=vsb.set) vsb.pack(side="right", fill="y") def _browse(self): d = filedialog.askdirectory(title="选择通达信安装目录（含 T0002）") if d: self.tdx_var.set(d) # ---- 启动 / 停止 ---- def _start(self): if self.scanning: return # 校验输入 try: sec_days = int(self.sec_spin.get()) stk_days = int(self.stk_spin.get()) top_n = int(self.top_n.get()) max_sec = int(self.max_sec.get()) except ValueError: messagebox.showerror("参数错误", "日数/N/板块数 必须为整数。") return if sec_days < 1 or stk_days < 1: messagebox.showerror("参数错误", "日数必须 ≥ 1。") return self.scanning = True self.start_btn.config(state="disabled") self.stop_btn.config(state="normal") self.tree.delete(*self.tree.get_children()) self.pb.config(value=0) params = { "tdx_path": self.tdx_var.get().strip() or None, "lookback_sector": sec_days, "lookback_stock": stk_days, "use_close": self.use_close.get(), "top_n": top_n, "max_sectors": max_sec, "member_cap": 300, } self.worker = ScanWorker(self.q, params) self.worker.start() def _stop(self): self.scanning = False self.status.config(text="已请求停止（当前批次完成后生效）。") self.start_btn.config(state="normal") self.stop_btn.config(state="disabled") # ---- 主线程轮询 queue，更新 UI ---- def _poll(self): try: while True: msg = self.q.get_nowait() self._handle(msg) except queue.Empty: pass self.root.after(80, self._poll) def _handle(self, msg): kind = msg[0] if kind == "status": self.status.config(text=msg[1]) elif kind == "phase": _, phase, total, _ = msg self._phase = phase self._total = total if phase == "sector": self.status.config(text="阶段 1/2：扫描板块创新高...") else: self.status.config(text="阶段 2/2：扫描板块内成分股...") elif kind == "prog": # 第一层进度 done, total, found = msg[1], msg[2], msg[3] self.status.config(text=f"扫描板块 {done}/{total}，已发现新高 {found} 个") if total: self.pb.config(value=int(done / total * 70)) elif kind == "prog2": # 第二层进度 m_done, sum_m, sname, sdone, stotal = msg[1], msg[2], msg[3], msg[4], msg[5] self.status.config(text=f"板块【{sname}】成分股 {sdone}/{stotal} | 总进度 {m_done}/{sum_m}") if sum_m: self.pb.config(value=int(70 + m_done / sum_m * 30)) elif kind == "done": report, t0 = msg[1], msg[2] self._fill_tree(report) self.pb.config(value=100) self.status.config(text=f"完成：共 {len(report)} 个创新高板块，耗时 {time.time()-t0:.1f}s") self.scanning = False self.start_btn.config(state="normal") self.stop_btn.config(state="disabled") elif kind == "error": messagebox.showerror("扫描出错", msg[1]) self.status.config(text="出错：" + msg[1]) self.scanning = False self.start_btn.config(state="normal") self.stop_btn.config(state="disabled") def _fill_tree(self, report): self.tree.delete(*self.tree.get_children()) if not report: self.tree.insert("", "end", text="结果", values=("当前没有板块创出新高。", "", "", "", "")) return for sec in report: siid = f"sec_{sec['code']}" self.tree.insert("", "end", iid=siid, text=sec["name"], values=(sec["name"], sec["code"], "—", f"+{sec['breakout_pct']:.2f}%", f"{sec.get('stock_count',0)}/{sec.get('member_total',0)} 只"), tags=("sector",)) for s in sec["stocks"]: nm = s["name"] or s["full_code"] self.tree.insert(siid, "end", text=nm, values=(nm, s["full_code"], f"{s['last']:.2f}", f"+{s['breakout_pct']:.2f}%", ""), tags=("stock",)) if not sec["stocks"]: self.tree.insert(siid, "end", text="（无成分股同步新高）", values=("（无成分股同步新高）", "", "", "", ""), tags=("stock",))def main(): ap = argparse.ArgumentParser(description="板块创新高扫描器 GUI") ap.add_argument("--tdx-path", type=str, default=None, help="通达信安装目录（含 T0002）") args = ap.parse_args() # eltdx 可用性检查 try: from eltdx import TdxClient # noqa: F401 except ImportError: root = tk.Tk() root.withdraw() messagebox.showerror("依赖缺失", "未找到 eltdx 库。请先安装：\n\n" " pip install eltdx\n\n" "或在 venv 环境中运行本程序。") return root = tk.Tk() App(root, args.tdx_path) root.mainloop()if __name__ == "__main__": main()
|
|手机版|小黑屋|
股指标网
( 渝ICP备2024026571号-1 )
发表于