dekun
38a38cb51d
页面先渲染三卡片，推荐表并行拉行情，持仓与推荐分别通过 API 加载。 Co-authored-by: Cursor <cursoragent@cursor.com>
100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
"""按账户资金推荐可交易品种（期货核心筛选）。"""
|
|
from __future__ import annotations
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from typing import Callable, Optional
|
|
|
|
from contract_specs import get_contract_spec
|
|
from symbols import PRODUCTS
|
|
|
|
|
|
def _letters_from_ths(ths_code: str) -> str:
|
|
import re
|
|
m = re.match(r"^([A-Za-z]+)", (ths_code or "").strip())
|
|
return m.group(1) if m else ""
|
|
|
|
|
|
def assess_product_for_capital(
|
|
product: dict,
|
|
capital: float,
|
|
price: Optional[float],
|
|
*,
|
|
max_position_pct: float = 50.0,
|
|
default_stop_ticks: int = 20,
|
|
) -> dict:
|
|
"""评估单品种在当前资金下是否可交易。"""
|
|
ths = product.get("ths") or ""
|
|
name = product.get("name") or ths
|
|
exchange = product.get("exchange") or ""
|
|
spec = get_contract_spec(ths + "8888")
|
|
mult = spec["mult"]
|
|
margin_rate = spec["margin_rate"]
|
|
tick = float(spec.get("tick_size") or 1.0)
|
|
p = float(price) if price and price > 0 else 0.0
|
|
cap = float(capital or 0)
|
|
|
|
if p <= 0:
|
|
return {
|
|
"ths": ths,
|
|
"name": name,
|
|
"exchange": exchange,
|
|
"status": "no_price",
|
|
"status_label": "暂无行情",
|
|
"min_capital_one_lot": None,
|
|
"margin_one_lot": None,
|
|
"risk_one_lot_1pct": None,
|
|
}
|
|
|
|
margin_one = p * mult * margin_rate
|
|
min_capital = margin_one / (max_position_pct / 100.0) if max_position_pct > 0 else margin_one
|
|
stop_dist = tick * default_stop_ticks
|
|
risk_one_lot = stop_dist * mult
|
|
risk_pct_1lot = (risk_one_lot / cap * 100) if cap > 0 else 999.0
|
|
|
|
can_margin = cap >= min_capital
|
|
can_risk = cap > 0 and risk_one_lot <= cap * 0.01
|
|
|
|
if can_margin and can_risk:
|
|
status, label = "ok", "推荐"
|
|
elif can_margin:
|
|
status, label = "margin_ok", "可开1手·止损偏宽"
|
|
else:
|
|
status, label = "blocked", "资金不足"
|
|
|
|
return {
|
|
"ths": ths,
|
|
"name": name,
|
|
"exchange": exchange,
|
|
"price": round(p, 4),
|
|
"mult": mult,
|
|
"tick_size": tick,
|
|
"margin_one_lot": round(margin_one, 2),
|
|
"min_capital_one_lot": round(min_capital, 2),
|
|
"risk_one_lot_1pct": round(risk_one_lot, 2),
|
|
"risk_pct_1lot_at_1pct_rule": round(risk_pct_1lot, 2),
|
|
"status": status,
|
|
"status_label": label,
|
|
}
|
|
|
|
|
|
def list_product_recommendations(
|
|
capital: float,
|
|
price_fn: Callable[[str], Optional[float]],
|
|
*,
|
|
max_position_pct: float = 50.0,
|
|
) -> list[dict]:
|
|
"""扫描全部品种并排序：推荐 > 可开 > 不足。"""
|
|
|
|
def _one(product: dict) -> dict:
|
|
ths = product["ths"]
|
|
main_code = price_fn(ths)
|
|
return assess_product_for_capital(
|
|
product, capital, main_code, max_position_pct=max_position_pct
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=10) as pool:
|
|
rows = list(pool.map(_one, PRODUCTS))
|
|
order = {"ok": 0, "margin_ok": 1, "blocked": 2, "no_price": 3}
|
|
rows.sort(key=lambda r: (order.get(r["status"], 9), r.get("min_capital_one_lot") or 1e18))
|
|
return rows
|