125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
def build_theme_library(
|
|
trade_date: str,
|
|
directory_rows: tuple[dict[str, Any], ...],
|
|
daily_rows: tuple[dict[str, Any], ...],
|
|
hot_rows: tuple[dict[str, Any], ...],
|
|
) -> dict[str, Any]:
|
|
daily = {str(row.get("ts_code") or ""): row for row in daily_rows}
|
|
hot = {
|
|
str(row.get("ts_code") or ""): int(_number(row.get("rank"), 9999))
|
|
for row in hot_rows
|
|
if str(row.get("data_type") or "") == "概念板块"
|
|
}
|
|
items = []
|
|
for row in directory_rows:
|
|
if str(row.get("type") or "").upper() != "N":
|
|
continue
|
|
if str(row.get("exchange") or "").upper() != "A":
|
|
continue
|
|
code = str(row.get("ts_code") or "")
|
|
name = str(row.get("name") or "").strip()
|
|
if not code or not name:
|
|
continue
|
|
quote = daily.get(code)
|
|
items.append(
|
|
{
|
|
"code": code,
|
|
"name": name,
|
|
"member_count": int(_number(row.get("count"))),
|
|
"change": round(_number((quote or {}).get("pct_change")), 2)
|
|
if quote
|
|
else None,
|
|
"close": round(_number((quote or {}).get("close")), 3) if quote else None,
|
|
"turnover_rate": round(_number((quote or {}).get("turnover_rate")), 2)
|
|
if quote
|
|
else None,
|
|
"hot_rank": hot.get(code),
|
|
"has_quote": bool(quote),
|
|
}
|
|
)
|
|
items.sort(
|
|
key=lambda item: (
|
|
bool(item["has_quote"]),
|
|
item["hot_rank"] is not None,
|
|
-(item["hot_rank"] or 9999),
|
|
_number(item["change"], -999),
|
|
),
|
|
reverse=True,
|
|
)
|
|
quoted = [item for item in items if item["has_quote"]]
|
|
return {
|
|
"trade_date": trade_date,
|
|
"summary": {
|
|
"theme_count": len(items),
|
|
"quoted_count": len(quoted),
|
|
"up_count": sum(_number(item["change"]) > 0 for item in quoted),
|
|
"down_count": sum(_number(item["change"]) < 0 for item in quoted),
|
|
"hot_count": len(hot),
|
|
},
|
|
"items": items,
|
|
}
|
|
|
|
|
|
def build_theme_detail(
|
|
trade_date: str,
|
|
theme: dict[str, Any],
|
|
member_rows: tuple[dict[str, Any], ...],
|
|
daily_rows: tuple[dict[str, Any], ...],
|
|
) -> dict[str, Any]:
|
|
daily = {str(row.get("ts_code") or ""): row for row in daily_rows}
|
|
members = []
|
|
seen = set()
|
|
for row in member_rows:
|
|
identifier = str(row.get("con_code") or "")
|
|
if not identifier or identifier in seen:
|
|
continue
|
|
seen.add(identifier)
|
|
quote = daily.get(identifier)
|
|
members.append(
|
|
{
|
|
"identifier": identifier,
|
|
"code": identifier.split(".")[0],
|
|
"name": str(row.get("con_name") or ""),
|
|
"change": round(_number((quote or {}).get("pct_chg")), 2)
|
|
if quote
|
|
else None,
|
|
"close": round(_number((quote or {}).get("close")), 2) if quote else None,
|
|
"amount": _number((quote or {}).get("amount")) * 1000 if quote else None,
|
|
"quoted": bool(quote),
|
|
}
|
|
)
|
|
members.sort(
|
|
key=lambda item: (
|
|
bool(item["quoted"]),
|
|
_number(item["change"], -999),
|
|
_number(item["amount"]),
|
|
),
|
|
reverse=True,
|
|
)
|
|
quoted = [item for item in members if item["quoted"]]
|
|
return {
|
|
"trade_date": trade_date,
|
|
"theme": theme,
|
|
"summary": {
|
|
"member_count": len(members),
|
|
"quoted_count": len(quoted),
|
|
"up_count": sum(_number(item["change"]) > 0 for item in quoted),
|
|
"down_count": sum(_number(item["change"]) < 0 for item in quoted),
|
|
"turnover_rate": theme.get("turnover_rate"),
|
|
},
|
|
"members": members,
|
|
}
|
|
|
|
|
|
def _number(value: Any, default: float = 0.0) -> float:
|
|
try:
|
|
number = float(value)
|
|
return number if number == number else default
|
|
except (TypeError, ValueError):
|
|
return default
|