QueryLucky Python Skill

简介

QueryLucky-Python 是 QueryLucky 期权扫描引擎的 Python 移植版，保持相同的核心逻辑：全品种实时扫描、双信号判定（IV分位 + IV-HV差）、8大策略类型、组合保证金白名单、2026/2027双年交易日历。适合集成到 Python 量化框架（vn.py、Qlib、自研回测）或作为独立脚本运行。

核心能力（与 Java 版一致）

1. 全品种覆盖（20个）

ag, ru, cu, al, zn, sn, ao, sp, ni, p,
lc, si, ps, sc, rb, i, lh, jm, a, b

2. 数据源（免费新浪接口）

• 期权链：http://stock.finance.sina.com.cn/futures/api/openapi.php/OptionService.getOptionData?type=futures&product={p}&exchange={e}&pinzhong={c}
• 期权实时：https://hq.sinajs.cn/list=P_OP_{code}
• 期货实时：https://hq.sinajs.cn/list=nf_{code}
• 期货日K：https://stock.finance.sina.com.cn/futures/api/json_v2.php/InnerFuturesNewService.getDailyKLine?symbol={symbol}
• 期权日K：https://stock.finance.sina.com.cn/futures/api/json_v2.php/FutureOptionAllService.getOptionDayline?symbol={symbol}

3. 双信号判定逻辑

# 卖方占优
sell_fav = (iv_pct >= 50 and iv_hv_diff >= 0) or (iv_pct < 0 and iv_hv_diff >= 3.0)

# 买方占优  
buy_fav = (iv_pct < 20 and iv_hv_diff <= 0) or (iv_pct < 0 and iv_hv_diff <= -3.0)

# 冲突→中性

4. 策略级阈值

5. 组合保证金白名单

PORTFOLIO_MARGIN_OK = {"au", "rb", "i", "p", "a", "b"}

6. 2026/2027 节假日表

内置法定节假日，trading_days_remaining(from_date, to_date) 精确计算剩余交易日。

Python 实现骨架

# query_lucky.py
"""
QueryLucky Python 版 - 全品种期权策略扫描引擎
依赖: Python 3.8+, requests, numpy, pandas, scipy(可选，Black-76用)
"""
from __future__ import annotations
import requests
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import date, datetime, timedelta
from math import log, sqrt, exp
from scipy.stats import norm  # 需安装 scipy

# ==================== 常量 ====================
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
REFERER = "https://finance.sina.com.cn"

PRODUCT_MAP = {
    "ru": ("ru_o", "shfe"), "cu": ("cu_o", "shfe"), "au": ("au_o", "shfe"),
    "al": ("al_o", "shfe"), "zn": ("zn_o", "shfe"), "ag": ("ag_o", "shfe"),
    "ni": ("ni_o", "shfe"), "rb": ("rb_o", "shfe"), "sn": ("sn_o", "shfe"),
    "ao": ("ao_o", "shfe"), "sp": ("sp_o", "shfe"), "sc": ("sc_o", "ine"),
    "a": ("a_o", "dce"), "b": ("b_o", "dce"), "m": ("m_o", "dce"),
    "c": ("c_o", "dce"), "i": ("i_o", "dce"), "y": ("y_o", "dce"),
    "p": ("p_o", "dce"), "v": ("v_o", "dce"), "l": ("l_o", "dce"),
    "pp": ("pp_o", "dce"), "eg": ("eg_o", "dce"), "eb": ("eb_o", "dce"),
    "lh": ("lh_o", "dce"), "jm": ("jm_o", "dce"),
    "lc": ("lc_o", "gfex"), "si": ("si_o", "gfex"), "ps": ("ps_o", "gfex"),
}

OPTION_PRODUCTS = {
    "ag": "沪银", "ru": "橡胶", "cu": "沪铜", "al": "沪铝", "zn": "沪锌",
    "sn": "沪锡", "ao": "氧化铝", "sp": "纸浆", "ni": "沪镍", "p": "棕榈油",
    "lc": "碳酸锂", "si": "工业硅", "ps": "多晶硅", "sc": "原油",
    "rb": "螺纹钢", "i": "铁矿石", "lh": "生猪", "jm": "焦煤",
    "a": "豆一", "b": "豆二",
}

SHFE_PRODUCTS = {"ru", "cu", "au", "al", "zn", "ag", "ni", "rb", "sn", "ao", "sp"}
GFEX_PRODUCTS = {"lc", "si", "ps"}
INE_PRODUCTS = {"sc"}

PORTFOLIO_MARGIN_OK = {"au", "rb", "i", "p", "a", "b"}

# 2026/2027 节假日 (月, 日)
HOLIDAYS_2026 = [(1,1),(1,2), (2,16),(2,17),(2,18),(2,19),(2,20),(2,23),
                 (4,6), (5,1),(5,4),(5,5), (6,19), (9,25),
                 (10,1),(10,2),(10,5),(10,6),(10,7)]
HOLIDAYS_2027 = [(1,1), (2,6),(2,7),(2,8),(2,9),(2,10),(2,11),(2,12),
                 (4,5), (5,1),(5,2),(5,3), (6,9),
                 (9,25),(9,26),(9,27),(9,28),(9,29),(9,30), (10,1)]
HOLIDAYS_BY_YEAR = {2026: HOLIDAYS_2026, 2027: HOLIDAYS_2027}

# ==================== 数据结构 ====================
@dataclass
class Quote:
    code: str
    strike: float
    is_call: bool
    bid: float
    ask: float
    mid: float
    volume: int
    open_interest: int
    last_price: float
    implied_vol: float = 0.0
    delta: float = 0.0
    gamma: float = 0.0
    theta: float = 0.0
    vega: float = 0.0

@dataclass
class Leg:
    action: str          # "买入"/"卖出"
    code: str
    strike: float
    price: float
    qty: int
    is_call: bool
    far: bool = False
    is_put: bool = False

@dataclass
class Strategy:
    name: str
    reason: str
    legs: List[Leg] = field(default_factory=list)
    pnl: str = ""
    greeks: str = ""
    score: float = 0.0
    handle_rank: int = 0
    pnl_arr: np.ndarray = field(default_factory=lambda: np.zeros(6))

@dataclass
class ProductScan:
    product: str
    name: str
    fut_price: float = 0.0
    iv: float = 0.0
    rv20: float = 0.0
    rv10: float = 0.0
    rv60: float = 0.0
    trend_pct: float = 0.0
    iv_percentile: float = -1.0
    high20: float = 0.0
    low20: float = 0.0
    high60: float = 0.0
    low60: float = 0.0
    verdict: str = "中性"
    error: Optional[str] = None
    strategies: List[Strategy] = field(default_factory=list)
    # 内部字段
    contract: str = ""
    far_contract: str = ""
    call_chain: List[Quote] = field(default_factory=list)
    put_chain: List[Quote] = field(default_factory=list)
    far_call_code: str = ""
    far_put_code: str = ""
    far_strike: float = 0.0
    far_call_mid: float = 0.0
    far_put_mid: float = 0.0

# ==================== 工具函数 ====================
def is_holiday(d: date) -> bool:
    holidays = HOLIDAYS_BY_YEAR.get(d.year, [])
    return (d.month, d.day) in holidays

def is_trading_day(d: date) -> bool:
    if d.weekday() >= 5:  # 5=Sat, 6=Sun
        return False
    return not is_holiday(d)

def trading_days_remaining(from_date: date, to_date: date) -> int:
    n = 0
    d = from_date + timedelta(days=1)
    while d <= to_date:
        if is_trading_day(d):
            n += 1
        d += timedelta(days=1)
    return n

def black76_price(f: float, k: float, t: float, v: float, is_call: bool) -> float:
    """Black-76 期权定价"""
    if t <= 0 or v <= 0:
        return max(0, f - k) if is_call else max(0, k - f)
    d1 = (log(f/k) + 0.5*v*v*t) / (v*sqrt(t))
    d2 = d1 - v*sqrt(t)
    if is_call:
        return f*norm.cdf(d1) - k*norm.cdf(d2)
    return k*norm.cdf(-d2) - f*norm.cdf(-d1)

def black76_greeks(f: float, k: float, t: float, v: float, is_call: bool) -> Tuple[float,float,float]:
    """返回 (delta, theta_1day, vega_1pct)"""
    if t <= 0 or v <= 0:
        itm = 1 if f > k else (-1 if f < k else 0)
        delta = 0.5 if itm == 0 else (1 if (is_call and itm > 0) or (not is_call and itm < 0) else 0)
        return delta * (1 if is_call else -1), 0.0, 0.0
    d1 = (log(f/k) + 0.5*v*v*t) / (v*sqrt(t))
    d2 = d1 - v*sqrt(t)
    nd1 = norm.pdf(d1)
    if is_call:
        delta = norm.cdf(d1)
        theta = -(f*v*nd1)/(2*sqrt(t))  # 年化
    else:
        delta = norm.cdf(d1) - 1
        theta = -(f*v*nd1)/(2*sqrt(t))
    vega = f * sqrt(t) * nd1 * 0.01  # 每1%波动率
    return delta, theta/252, vega  # theta转日

def implied_volatility(price: float, f: float, k: float, t: float, is_call: bool) -> float:
    """二分法求隐含波动率"""
    if price <= 0 or t <= 0:
        return 0.0
    lo, hi = 0.001, 5.0
    for _ in range(50):
        mid = (lo + hi) / 2
        pv = black76_price(f, k, t, mid, is_call)
        if pv > price:
            hi = mid
        else:
            lo = mid
    return (lo + hi) / 2

# ==================== 核心类 ====================
class QueryLucky:
    def __init__(self, session: Optional[requests.Session] = None):
        self.session = session or requests.Session()
        self.session.headers.update({"User-Agent": UA, "Referer": REFERER})

    # ----- HTTP 请求 -----
    def _get(self, url: str) -> Optional[dict]:
        try:
            r = self.session.get(url, timeout=10)
            r.raise_for_status()
            return r.json()
        except Exception:
            return None

    def _get_text(self, url: str) -> str:
        try:
            r = self.session.get(url, timeout=10)
            r.raise_for_status()
            return r.text
        except Exception:
            return ""

    # ----- 数据获取 -----
    def fetch_option_chain(self, product: str) -> Optional[dict]:
        pcode, exch = PRODUCT_MAP[product]
        url = f"http://stock.finance.sina.com.cn/futures/api/openapi.php/OptionService.getOptionData?type=futures&product={pcode}&exchange={exch}&pinzhong={product}"
        return self._get(url)

    def fetch_futures_quote(self, product: str) -> float:
        """返回期货最新价"""
        url = f"https://hq.sinajs.cn/list=nf_{product}"
        txt = self._get_text(url)
        if not txt or "=" not in txt:
            return 0.0
        parts = txt.split('"')[1].split(",")
        return float(parts[0]) if parts else 0.0

    def fetch_option_quotes(self, codes: List[str]) -> Dict[str, dict]:
        """批量获取期权实时行情，返回 {code: {bid,ask,last,vol,oi}}"""
        if not codes:
            return {}
        url = "https://hq.sinajs.cn/list=" + ",".join(f"P_OP_{c}" for c in codes)
        txt = self._get_text(url)
        out = {}
        for line in txt.strip().split("\n"):
            if "=" not in line:
                continue
            code = line.split("=")[0].split("_")[-1].strip('"')
            vals = line.split('"')[1].split(",")
            if len(vals) >= 10:
                out[code] = {
                    "bid": float(vals[2]) if vals[2] else 0.0,
                    "ask": float(vals[3]) if vals[3] else 0.0,
                    "last": float(vals[0]) if vals[0] else 0.0,
                    "vol": int(vals[8]) if vals[8] else 0,
                    "oi": int(vals[9]) if vals[9] else 0,
                }
        return out

    def fetch_futures_kline(self, symbol: str, count: int = 120) -> List[List[float]]:
        """返回 [[date,open,high,low,close,vol], ...] 最近count根"""
        url = f"https://stock.finance.sina.com.cn/futures/api/json_v2.php/InnerFuturesNewService.getDailyKLine?symbol={symbol}"
        data = self._get(url)
        if not data:
            return []
        # 新浪返回倒序，需反转
        bars = []
        for item in data[:count]:
            try:
                d = datetime.strptime(item["date"], "%Y-%m-%d").date()
                bars.append([d, float(item["open"]), float(item["high"]),
                            float(item["low"]), float(item["close"]), float(item["volume"])])
            except:
                pass
        bars.reverse()  # 正序
        return bars

    def fetch_option_dayline(self, code: str, count: int = 60) -> List[float]:
        """返回期权日线收盘价列表(最近count根)"""
        url = f"https://stock.finance.sina.com.cn/futures/api/json_v2.php/FutureOptionAllService.getOptionDayline?symbol={code}"
        data = self._get(url)
        if not data:
            return []
        closes = []
        for item in data[:count]:
            try:
                closes.append(float(item["close"]))
            except:
                pass
        closes.reverse()
        return closes

    # ----- 扫描单品种 -----
    def scan_product(self, product: str) -> ProductScan:
        r = ProductScan(product=product, name=OPTION_PRODUCTS.get(product, product))
        try:
            # 1. 期货现价
            spot = self.fetch_futures_quote(product)
            if spot <= 0:
                r.error = "期货行情为空"
                return r
            r.fut_price = spot

            # 2. 期权链
            chain_data = self.fetch_option_chain(product)
            if not chain_data or "result" not in chain_data or "data" not in chain_data["result"]:
                r.error = "期权链为空"
                return r

            # 3. 解析合约月份，选主力（持仓量最大、距到期>10天）
            # 简化：取最近月份
            # 实际需解析 chain_data["result"]["data"]["contractMonth"] 等字段
            # 这里略过详细解析，假设已得到主力合约代码 r.contract
            # 并构建 call_chain / put_chain (List[Quote])

            # 4. 计算 RV10/20/60
            bars = self.fetch_futures_kline(product + "2412", 120)  # 示例主力合约
            if len(bars) >= 20:
                closes = np.array([b[4] for b in bars])
                rets = np.diff(np.log(closes))
                r.rv20 = float(np.std(rets[-20:]) * np.sqrt(252) * 100)
                r.rv10 = float(np.std(rets[-10:]) * np.sqrt(252) * 100) if len(rets) >= 10 else 0
                r.rv60 = float(np.std(rets[-60:]) * np.sqrt(252) * 100) if len(rets) >= 60 else 0
                highs = np.array([b[2] for b in bars])
                lows = np.array([b[3] for b in bars])
                r.high20, r.low20 = float(highs[-20:].max()), float(lows[-20:].min())
                r.high60, r.low60 = float(highs[-60:].max()), float(lows[-60:].min())

            # 5. 计算 ATM IV 序列 & 分位
            # 这里简化：假设已得到 iv_series (List[float])
            # iv_series = self._calc_iv_series(r, spot)
            # r.iv = iv_series[-1] if iv_series else 0
            # if len(iv_series) >= 20:
            #     r.iv_percentile = (np.array(iv_series) < r.iv).mean() * 100

            # 6. 趋势
            if len(bars) >= 10:
                r.trend_pct = (bars[-1][4] - bars[-10][4]) / bars[-10][4] * 100

            # 7. 构建策略
            if r.error is None:
                self.build_strategies(r, spot)

        except Exception as e:
            r.error = str(e)
        return r

    def scan_all(self) -> List[ProductScan]:
        results = []
        for p in OPTION_PRODUCTS:
            results.append(self.scan_product(p))
        return results

    # ----- 策略构建（核心逻辑，同 Java 版） -----
    def build_strategies(self, r: ProductScan, spot: float):
        if r.iv <= 0 or r.fut_price <= 0:
            return

        diff = r.iv - r.rv20
        iv_pct = r.iv_percentile if r.iv_percentile >= 0 else -1.0
        diff_txt = f"{diff:+.1f}"
        iv_txt = f"IV分位{iv_pct:.0f}%" if iv_pct >= 0 else "IV分位NA"
        trend_txt = f"近10日{r.trend_pct:+.1f}%"

        sell_fav = (iv_pct >= 0 and iv_pct >= 50 and diff >= 0) or (iv_pct < 0 and diff >= 3.0)
        buy_fav = (iv_pct >= 0 and iv_pct < 20 and diff <= 0) or (iv_pct < 0 and diff <= -3.0)

        sell_env = f"IV中高位({iv_txt}), IV-HV20={diff_txt}pp" + (" 显著偏贵" if diff >= 3.0 else " 偏贵")
        buy_env = f"IV历史低位({iv_txt}), IV-HV20={diff_txt}pp" + (" 显著低估" if diff <= -3.0 else " 低估")

        # 这里需要完整的期权链 quote 对象，简化演示只生成框架
        # 实际需实现：ATM定位、pickStrangleLeg、希腊字母、PnL网格、评分排序

        # 示例：卖方分支
        if sell_fav:
            if diff >= 3.0:
                self._add_strat(r, "卖出跨式(Sell Straddle)",
                    f"{sell_env}, 双卖ATM收割时间价值与波动率回落; 止损: 突破平衡点或IV飙升; 仓位≤20%",
                    spot)
            if diff >= 1.5:
                self._add_strat(r, "卖出宽跨式(Sell Strangle)",
                    f"{sell_env}, Delta≈0.30选档覆盖60日区间; 净Delta中性; 止损: 突破卖出档或IV飙升",
                    spot)
            self._add_strat(r, "铁鹰(Iron Condor)",
                f"IV中高位({iv_txt})预期波动率回落窄幅震荡, 卖±1档买±2档, 风险有限; 止损: 突破卖出档",
                spot)

        elif buy_fav:
            self._add_strat(r, "买入跨式(Long Straddle)",
                f"{buy_env}, 变盘窗口提前1-5天布局; 最大亏损=权利金, 盈亏比极高", spot)
            self._add_strat(r, "买入宽跨式(Long Strangle)",
                f"{buy_env}, 虚值双买低成本博变盘; 最大亏损=权利金", spot)

        else:
            self._add_strat(r, "卖出跨式(轻仓)",
                "横盘假设+IV中性, 小仓位收theta, 严格止损(突破平衡点); 仓位≤20%", spot)

        # 排序取Top3
        r.strategies.sort(key=lambda s: (-s.handle_rank, -s.score))
        if len(r.strategies) > 3:
            r.strategies = r.strategies[:3]

    def _add_strat(self, r: ProductScan, name: str, reason: str, spot: float):
        st = Strategy(name=name, reason=reason)
        # 此处需根据策略添加 legs
        # 计算 PnL、Greeks、评分
        r.strategies.append(st)

    # ----- PnL / Greeks / 评分（同 Java 版逻辑） -----
    def _calc_pnl(self, st: Strategy, spot: float, t_years: float, far_iv: float) -> np.ndarray:
        # 网格法计算到期损益分布
        # 返回 [maxP, minP, up_unbounded, dn_unbounded, be1, be2]
        pass

    def _fmt_pnl(self, s: np.ndarray) -> str:
        sb = f"最大盈 {s[0]:.0f}"
        if s[3] > 0:
            sb += " | 最大亏 无限"
        else:
            sb += f" | 最大亏 {s[1]:.0f}"
        if s[4] > 0:
            sb += f" | 平衡 {s[4]:.0f}"
            if s[5] > 0:
                sb += f"/{s[5]:.0f}"
        if s[2] > 0:
            sb += " | 盈利无上界"
        if s[3] > 0:
            sb += " | 亏损无上界"
        return sb

    def _score(self, s: np.ndarray) -> float:
        if s[0] <= 0:
            return -1.0
        rr = s[0] / (-s[1]) if s[1] < 0 else 9.9
        if s[3] > 0:
            rr *= 0.4
        if s[2] > 0:
            rr *= 1.3
        return rr

# ==================== 使用示例 ====================
if __name__ == "__main__":
    ql = QueryLucky()
    scans = ql.scan_all()
    for r in scans:
        if r.error:
            print(f"{r.product:4s} ERROR {r.error}")
            continue
        iv_pct = f"{r.iv_percentile:.0f}" if r.iv_percentile >= 0 else "NA"
        print(f"{r.product:4s} spot={r.fut_price:.0f} IV={r.iv:.1f}%(pct={iv_pct}) HV20={r.rv20:.1f}% diff={r.iv-r.rv20:+.1f}pp verdict={r.verdict}")
        for st in r.strategies:
            print(f"   [{st.handle_rank:.0f}] {st.name}")
            print(f"      {st.reason}")
            print(f"      PnL: {st.pnl}")

依赖安装

pip install requests numpy pandas scipy

与 Java 版的差异

迁移清单

• [x] 常量与品种映射
• [x] 节假日表 2026/2027
• [x] 交易日计算
• [x] Black-76 定价/希腊字母/IV求解
• [x] 双信号判定 + 策略级阈值
• [x] 组合保证金白名单
• [ ] 期权链解析完整逻辑
• [ ] ATM IV 历史序列反推（需期权日K）
• [ ] pickStrangleLeg（Delta 0.30 选档+区间覆盖）
• [ ] PnL 网格计算 + 无限风险标注
• [ ] evaluateProbability（胜率/行权/触及概率）
• [ ] 并行扫描优化

快速验证

python query_lucky.py

预期输出格式与 Java 版一致：每品种 verdict + Top3 策略名、理由、PnL。