"""区间内 Volume Profile。""" from __future__ import annotations from typing import Any, Dict, List import numpy as np import pandas as pd def compute_volume_profile( df: pd.DataFrame, start_idx: int, end_idx: int, bin_count: int = 50, value_area_pct: float = 0.70, ) -> Dict[str, Any]: seg = df.iloc[start_idx : end_idx + 1] if seg.empty: return {"bins": [], "poc": None, "vah": None, "val": None, "bin_count": bin_count} typical = (seg["high"].astype(float) + seg["low"].astype(float) + seg["close"].astype(float)) / 3.0 vol = seg["volume"].astype(float).fillna(0.0) lo = float(seg["low"].min()) hi = float(seg["high"].max()) if not np.isfinite(lo) or not np.isfinite(hi) or hi <= lo: mid = float(seg["close"].iloc[-1]) return { "bins": [{"price": mid, "volume": float(vol.sum())}], "poc": mid, "vah": mid, "val": mid, "bin_count": 1, } edges = np.linspace(lo, hi, bin_count + 1) # 右开最后一桶闭合 idx = np.clip(np.digitize(typical.values, edges) - 1, 0, bin_count - 1) vols = np.zeros(bin_count, dtype=float) for i, v in zip(idx, vol.values): vols[i] += float(v) centers = (edges[:-1] + edges[1:]) / 2.0 poc_i = int(np.argmax(vols)) if vols.sum() > 0 else bin_count // 2 poc = float(centers[poc_i]) # Value Area:从 POC 向两侧扩展直到累计 >= value_area_pct total = float(vols.sum()) or 1.0 target = total * value_area_pct left = right = poc_i acc = float(vols[poc_i]) while acc < target and (left > 0 or right < bin_count - 1): left_v = vols[left - 1] if left > 0 else -1.0 right_v = vols[right + 1] if right < bin_count - 1 else -1.0 if right_v >= left_v and right < bin_count - 1: right += 1 acc += float(vols[right]) elif left > 0: left -= 1 acc += float(vols[left]) else: break bins: List[Dict[str, float]] = [ {"price": float(centers[i]), "volume": float(vols[i])} for i in range(bin_count) ] return { "bins": bins, "poc": poc, "vah": float(centers[right]), "val": float(centers[left]), "bin_count": bin_count, }