#!/usr/bin/env python3 """ 高股息股票扫描器 - 港股/美股/ A股：全部走LongPort API - A股股息率：使用预设数据（定期更新） """ import os, sys, json from datetime import datetime # === 读取LongPort环境变量 === bashrc = open(os.path.expanduser("~/.bashrc")).read() for line in bashrc.splitlines(): if line.startswith("export LONGPORT_") or line.startswith("export LONGBRIDGE_"): parts = line.replace("export ", "").split("=", 1) if len(parts) == 2: os.environ[parts[0]] = parts[1].strip('"').strip("'") from longport import openapi from longport.openapi import CalcIndex cfg = openapi.Config.from_env() ctx = openapi.QuoteContext(config=cfg) # === A股候选池（股息率+派息频率，定期更新） === A_SHARE_POOL = { "601088.SH": {"name": "中国神华", "yield": 6.7, "freq": "年+中期", "sector": "煤炭"}, "601328.SH": {"name": "交通银行", "yield": 6.2, "freq": "年度", "sector": "银行"}, "601398.SH": {"name": "工商银行", "yield": 5.9, "freq": "年度", "sector": "银行"}, "601288.SH": {"name": "农业银行", "yield": 5.8, "freq": "年度", "sector": "银行"}, "601939.SH": {"name": "建设银行", "yield": 6.0, "freq": "年度", "sector": "银行"}, "601988.SH": {"name": "中国银行", "yield": 5.7, "freq": "年度", "sector": "银行"}, "600900.SH": {"name": "长江电力", "yield": 3.8, "freq": "年度", "sector": "电力"}, "601857.SH": {"name": "中国石油", "yield": 5.5, "freq": "年度", "sector": "能源"}, "600028.SH": {"name": "中国石化", "yield": 5.2, "freq": "年度", "sector": "能源"}, "601728.SH": {"name": "中国电信", "yield": 4.8, "freq": "年度", "sector": "电信"}, "600036.SH": {"name": "招商银行", "yield": 4.5, "freq": "年度", "sector": "银行"}, "601166.SH": {"name": "兴业银行", "yield": 5.8, "freq": "年度", "sector": "银行"}, "601818.SH": {"name": "光大银行", "yield": 5.9, "freq": "年度", "sector": "银行"}, "600377.SH": {"name": "宁沪高速", "yield": 6.2, "freq": "年度", "sector": "高速"}, "601666.SH": {"name": "平煤股份", "yield": 6.2, "freq": "年度", "sector": "煤炭"}, "600023.SH": {"name": "浙能电力", "yield": 5.5, "freq": "年度", "sector": "电力"}, "000858.SZ": {"name": "五粮液", "yield": 10.5, "freq": "年度", "sector": "白酒"}, "000568.SZ": {"name": "泸州老窖", "yield": 7.0, "freq": "年度", "sector": "白酒"}, "000937.SZ": {"name": "冀中能源", "yield": 11.0, "freq": "年度", "sector": "煤炭"}, "002304.SZ": {"name": "洋河股份", "yield": 10.8, "freq": "年度", "sector": "白酒"}, "000596.SZ": {"name": "古井贡酒", "yield": 6.9, "freq": "年度", "sector": "白酒"}, "000001.SZ": {"name": "平安银行", "yield": 5.4, "freq": "年度", "sector": "银行"}, "600519.SH": {"name": "贵州茅台", "yield": 5.0, "freq": "年+中期", "sector": "白酒"}, } # === 港股/美股候选池 === STOCK_POOLS = { "hk": { "flag": "🇭🇰", "label": "港股", "min_yield": 7.0, "tickers": [ "00939.HK", "01398.HK", "00857.HK", "00883.HK", "01088.HK", "00941.HK", "06823.HK", "00823.HK", "02191.HK", "00778.HK", "00405.HK", "02778.HK", "01898.HK", "03988.HK", "01288.HK", "00006.HK", "00002.HK", "00003.HK", "00012.HK", "00016.HK", "00388.HK", "02318.HK", "00027.HK", "01299.HK", ], "freq": { "00939.HK": "半年", "01398.HK": "半年", "00857.HK": "半年", "00883.HK": "半年", "01088.HK": "半年", "00941.HK": "半年", "06823.HK": "半年", "00823.HK": "半年", "02191.HK": "季度", "00778.HK": "半年", "00405.HK": "半年", "02778.HK": "半年", "01898.HK": "半年", "03988.HK": "半年", "01288.HK": "半年", "00006.HK": "半年", "00002.HK": "半年", "00003.HK": "半年", "00012.HK": "半年", "00016.HK": "半年", "00388.HK": "半年", "02318.HK": "半年", "00027.HK": "半年", "01299.HK": "半年", }, }, "us": { "flag": "🇺🇸", "label": "美股", "min_yield": 7.0, "tickers": [ "NLY.US", "HTGC.US", "ARCC.US", "AGNC.US", "PSEC.US", "TWO.US", "STWD.US", "OFS.US", "MAIN.US", "ABR.US", "CIM.US", "NYMT.US", "SBR.US", "PDI.US", "PTY.US", ], "freq": { "NLY.US": "季度", "HTGC.US": "季度", "ARCC.US": "季度", "AGNC.US": "月度", "PSEC.US": "月度", "TWO.US": "季度", "STWD.US": "季度", "OFS.US": "季度", "MAIN.US": "月度", "ABR.US": "季度", "CIM.US": "季度", "NYMT.US": "季度", "SBR.US": "月度", "PDI.US": "月度", "PTY.US": "季度", }, }, } def scan_with_longport(market_key): """用LongPort扫描港股/美股，获取实时价格+股息率""" pool = STOCK_POOLS[market_key] tickers = pool["tickers"] min_yield = pool["min_yield"] flag = pool["flag"] freq_map = pool["freq"] results = [] all_quotes = {} for i in range(0, len(tickers), 15): batch = tickers[i:i+15] try: quotes = ctx.quote(batch) for q in quotes: all_quotes[q.symbol] = { "price": float(q.last_done), "prev_close": float(q.prev_close), "name": q.name if hasattr(q, 'name') else q.symbol, } except: pass try: indexes = ctx.calc_indexes(batch, [CalcIndex.DividendYield]) for idx in indexes: if idx.symbol in all_quotes: dy = getattr(idx, 'dividend_yield', None) if dy is not None: all_quotes[idx.symbol]["yield"] = float(dy) * 100 except: pass for sym, info in all_quotes.items(): dy = info.get("yield", 0) if dy >= min_yield: change_pct = (info["price"] - info["prev_close"]) / info["prev_close"] * 100 results.append({ "symbol": sym, "name": info.get("name", sym), "price": info["price"], "change_pct": change_pct, "yield": dy, "freq": freq_map.get(sym, "未知"), "flag": flag, }) results.sort(key=lambda x: x["yield"], reverse=True) return results[:5] def scan_a_share(): """用LongPort获取A股实时价格，配合预设股息率数据""" flag = "🇨🇳" min_yield = 5.0 tickers = list(A_SHARE_POOL.keys()) all_quotes = {} # LongPort获取A股实时价格 for i in range(0, len(tickers), 15): batch = tickers[i:i+15] try: quotes = ctx.quote(batch) for q in quotes: all_quotes[q.symbol] = { "price": float(q.last_done), "prev_close": float(q.prev_close), "name": q.name if hasattr(q, 'name') else q.symbol, } except: pass results = [] for sym, info in A_SHARE_POOL.items(): if info["yield"] < min_yield: continue q = all_quotes.get(sym) if not q or q["price"] <= 0: continue change_pct = (q["price"] - q["prev_close"]) / q["prev_close"] * 100 results.append({ "symbol": sym, "name": q.get("name", info["name"]), "price": q["price"], "change_pct": change_pct, "yield": info["yield"], "freq": info["freq"], "flag": flag, "sector": info.get("sector", ""), }) results.sort(key=lambda x: x["yield"], reverse=True) return results[:10] def calc_ladder(price): return [ {"tier": 1, "pct": -3, "price": round(price * 0.97, 2)}, {"tier": 2, "pct": -6, "price": round(price * 0.94, 2)}, {"tier": 3, "pct": -10, "price": round(price * 0.90, 2)}, ] def format_result(label, flag, results, total_scanned): now = datetime.now().strftime("%Y-%m-%d %H:%M") if not results: return "" # 无候选时不输出，避免空推送 if "error" in results[0]: return f"{flag} {label}高息扫描 | {now}\n\n❌ {results[0]['error']}" lines = [f"{flag} {label}高息TOP | {now}", ""] for i, r in enumerate(results, 1): medal = ["🥇", "🥈", "🥉", "4️⃣", "5️⃣", "6️⃣", "7️⃣", "8️⃣", "9️⃣", "🔟"][min(i-1, 9)] chg = "📈" if r["change_pct"] >= 0 else "📉" sector = f" [{r['sector']}]" if r.get("sector") else "" lines.append(f"{medal} {r['symbol']}{sector}") lines.append(f" {r['name']}") lines.append(f" 💰 现价: {r['price']:.2f} {chg} {r['change_pct']:+.1f}%") lines.append(f" 📊 股息率: {r['yield']:.1f}% 派息: {r['freq']}") ladder = calc_ladder(r["price"]) lstr = " → ".join([f"T{l['tier']}:{l['price']:.2f}({l['pct']}%)" for l in ladder]) lines.append(f" 🪜 阶梯: {lstr}") lines.append("") lines.append("━━━━━━━━━━━━━") lines.append(f"📋 共扫描 {total_scanned} 只，筛出 {len(results)} 只") return "\n".join(lines) if __name__ == "__main__": market = sys.argv[1] if len(sys.argv) > 1 else "all" if market == "cn": results = scan_a_share() output = format_result("A股", "🇨🇳", results, len(A_SHARE_POOL)) if output.strip(): print(output) elif market == "hk": results = scan_with_longport("hk") output = format_result("港股", "🇭🇰", results, len(STOCK_POOLS["hk"]["tickers"])) if output.strip(): print(output) elif market == "us": results = scan_with_longport("us") output = format_result("美股", "🇺🇸", results, len(STOCK_POOLS["us"]["tickers"])) if output.strip(): print(output) elif market == "all": outputs = [] outputs.append(format_result("港股", "🇭🇰", scan_with_longport("hk"), len(STOCK_POOLS["hk"]["tickers"]))) outputs.append(format_result("美股", "🇺🇸", scan_with_longport("us"), len(STOCK_POOLS["us"]["tickers"]))) outputs.append(format_result("A股", "🇨🇳", scan_a_share(), len(A_SHARE_POOL))) # 过滤空结果 outputs = [o for o in outputs if o.strip()] if outputs: print("\n\n".join(outputs)) # 无候选时不输出任何内容 else: print(f"用法: python3 {sys.argv[0]} [hk|us|cn|all]")