ricequant/ricequant-skills/sector-overview

sector-overview/
├── SKILL.md
├── scripts/
│   └── generate_report.py
├── assets/
│   └── template.md
└── references/
    └── data_contract.md

REPORT_DATE="${REPORT_DATE:-$(date +%F)}"
INDUSTRY_NAME="${INDUSTRY_NAME:-白酒}"
INDUSTRY_LEVEL="${INDUSTRY_LEVEL:-third}"
INDUSTRY_VALUE="${INDUSTRY_VALUE:-白酒}"
INDUSTRY_SOURCE="${INDUSTRY_SOURCE:-citics_2019}"
BENCHMARK_ORDER_BOOK_ID="${BENCHMARK_ORDER_BOOK_ID:-000300.XSHG}"
PRICE_START_DATE="$(python3 - <<PY
from datetime import date, timedelta
report_date = date.fromisoformat("${REPORT_DATE}")
print((report_date - timedelta(days=180)).isoformat())
PY
)"
FACTOR_DATE="$(python3 - <<PY
import json
import subprocess
from datetime import date, timedelta
 
def has_non_empty_factor(day: str) -> bool:
    payload = json.dumps({
        "order_book_ids": ["600519.XSHG", "000858.XSHE"],
        "factor": "market_cap",
        "start_date": day,
        "end_date": day,
    }, ensure_ascii=False)
    rows = json.loads(subprocess.check_output(
        ["rqdata", "stock", "cn", "financial-indicator", "--payload", payload, "--format", "json"],
        text=True,
    ))
    return any(isinstance(row, dict) and row.get("market_cap") not in (None, "", "null") for row in rows)
 
cursor = date.fromisoformat("${REPORT_DATE}")
for _ in range(10):
    if has_non_empty_factor(cursor.isoformat()):
        print(cursor.isoformat())
        break
    cursor -= timedelta(days=1)
else:
    raise SystemExit("最近 10 个自然日都未找到非空 market_cap 因子日期")
PY
)"
HISTORY_START_QUARTER="$(python3 - <<PY
from datetime import date
report_date = date.fromisoformat("${REPORT_DATE}")
print(f"{report_date.year - 2}q1")
PY
)"
HISTORY_END_QUARTER="$(python3 - <<PY
from datetime import date
report_date = date.fromisoformat("${REPORT_DATE}")
print(f"{report_date.year}q4")
PY
)"
 
DATA_DIR="${DATA_DIR:-$HOME/rq_equities_reports/sector_overview}"
OUTPUT_MD="${OUTPUT_MD:-$DATA_DIR/sector_overview_${INDUSTRY_NAME}_${REPORT_DATE}.md}"

{
  "industry_name": "白酒",
  "industry_level": "third",
  "industry_value": "白酒",
  "industry_source": "citics_2019",
  "benchmark_order_book_id": "000300.XSHG"
}

mkdir -p "$DATA_DIR"
 
rqdata stock cn list --payload "{
  \"date\": \"$FACTOR_DATE\",
  \"type\": \"CS\"
}" --format json > "$DATA_DIR/stock_list.json"
 
python3 - "$DATA_DIR/stock_list.json" "$DATA_DIR/industry_map.json" "$DATA_DIR/stock_pool.json" <<'PY'
import json
import subprocess
import sys
from pathlib import Path
 
stock_list_path = Path(sys.argv[1])
industry_map_path = Path(sys.argv[2])
stock_pool_path = Path(sys.argv[3])
definition_path = stock_list_path.parent / "sector_definition.json"
definition = json.loads(definition_path.read_text()) if definition_path.exists() else {}
 
industry_level = definition.get("industry_level", "third")
industry_value = definition.get("industry_value")
industry_source = definition.get("industry_source", "citics_2019")
if not industry_value:
    raise SystemExit("sector_definition.json 缺少 industry_value")
 
payload = json.loads(stock_list_path.read_text())
items = payload if isinstance(payload, list) else payload.get("data", [])
order_book_ids = [item["order_book_id"] for item in items if isinstance(item, dict) and item.get("order_book_id")]
 
industry_rows = []
for start in range(0, len(order_book_ids), 800):
    chunk = order_book_ids[start:start + 800]
    cmd = [
        "rqdata", "stock", "cn", "industry",
        "--payload", json.dumps({
            "order_book_ids": chunk,
            "date": definition.get("factor_date"),
            "level": 0,
            "source": industry_source,
        }, ensure_ascii=False),
        "--format", "json",
    ]
    industry_rows.extend(json.loads(subprocess.check_output(cmd, text=True)))
 
industry_map_path.write_text(json.dumps(industry_rows, ensure_ascii=False, indent=2), encoding="utf-8")
 
field_map = {
    "first": "first_industry_name",
    "second": "second_industry_name",
    "third": "third_industry_name",
}
field_name = field_map[industry_level]
selected = [row for row in industry_rows if isinstance(row, dict) and row.get(field_name) == industry_value]
stock_pool_path.write_text(json.dumps(selected, ensure_ascii=False, indent=2), encoding="utf-8")
PY

STOCK_POOL_JSON="$(python3 - "$DATA_DIR/stock_pool.json" <<'PY'
import json
import sys
from pathlib import Path
payload = json.loads(Path(sys.argv[1]).read_text())
items = payload if isinstance(payload, list) else payload.get("data", [])
order_book_ids = []
for item in items:
    if isinstance(item, dict) and item.get("order_book_id"):
        order_book_ids.append(item["order_book_id"])
print(json.dumps(order_book_ids, ensure_ascii=False))
PY
)"
 
rqdata stock cn instruments --payload "{
  \"order_book_ids\": $STOCK_POOL_JSON
}" --format json > "$DATA_DIR/instrument_meta.json"
 
rqdata stock cn financial --payload "{
  \"order_book_ids\": $STOCK_POOL_JSON,
  \"fields\": [\"revenue\", \"net_profit\", \"gross_profit\", \"total_assets\", \"total_liabilities\", \"cash_from_operating_activities\"],
  \"start_quarter\": \"$HISTORY_START_QUARTER\",
  \"end_quarter\": \"$HISTORY_END_QUARTER\",
  \"statements\": \"all\"
}" --format json > "$DATA_DIR/historical_financials.json"
 
cp "$DATA_DIR/historical_financials.json" "$DATA_DIR/latest_financials.json"
 
rqdata stock cn financial-indicator --payload "{
  \"order_book_ids\": $STOCK_POOL_JSON,
  \"factor\": \"return_on_equity_weighted_average\",
  \"start_date\": \"$FACTOR_DATE\",
  \"end_date\": \"$FACTOR_DATE\"
}" --format json > "$DATA_DIR/roe.json"
 
rqdata stock cn financial-indicator --payload "{
  \"order_book_ids\": $STOCK_POOL_JSON,
  \"factor\": \"market_cap\",
  \"start_date\": \"$FACTOR_DATE\",
  \"end_date\": \"$FACTOR_DATE\"
}" --format json > "$DATA_DIR/market_cap.json"
 
rqdata stock cn financial-indicator --payload "{
  \"order_book_ids\": $STOCK_POOL_JSON,
  \"factor\": \"pe_ratio\",
  \"start_date\": \"$FACTOR_DATE\",
  \"end_date\": \"$FACTOR_DATE\"
}" --format json > "$DATA_DIR/pe_ratio.json"
 
rqdata stock cn financial-indicator --payload "{
  \"order_book_ids\": $STOCK_POOL_JSON,
  \"factor\": \"pb_ratio\",
  \"start_date\": \"$FACTOR_DATE\",
  \"end_date\": \"$FACTOR_DATE\"
}" --format json > "$DATA_DIR/pb_ratio.json"
 
rqdata stock cn financial-indicator --payload "{
  \"order_book_ids\": $STOCK_POOL_JSON,
  \"factor\": \"dividend_yield\",
  \"start_date\": \"$FACTOR_DATE\",
  \"end_date\": \"$FACTOR_DATE\"
}" --format json > "$DATA_DIR/dividend_yield.json"
 
rqdata stock cn price --payload "{
  \"order_book_ids\": $STOCK_POOL_JSON,
  \"start_date\": \"$PRICE_START_DATE\",
  \"end_date\": \"$FACTOR_DATE\",
  \"fields\": [\"close\"],
  \"adjust_type\": \"post\"
}" --format json > "$DATA_DIR/price_period.json"
 
rqdata index price --payload "{
  \"order_book_ids\": [\"$BENCHMARK_ORDER_BOOK_ID\"],
  \"start_date\": \"$PRICE_START_DATE\",
  \"end_date\": \"$FACTOR_DATE\",
  \"fields\": [\"close\"]
}" --format json > "$DATA_DIR/benchmark_price.json"

python3 sector-overview/scripts/generate_report.py \
  --industry "$INDUSTRY_NAME" \
  --data-dir "$DATA_DIR" \
  --report-date "$REPORT_DATE" \
  --output "$OUTPUT_MD"

python3 sector-overview/generate_report.py \
  --industry "$INDUSTRY_NAME" \
  --data-dir "$DATA_DIR" \
  --report-date "$REPORT_DATE" \
  --output "$OUTPUT_MD"