candidate_pivot_pos = -1
candidate_pivot_value_log = np.nan
+ def minmax_scale(values: list[float]) -> list[float]:
+ """Scale values to [0, 1] range preserving NaN.
+
+ Edge cases:
+ - Empty: return []
+ - All NaN: return all NaN
+ - All same finite value: return [0.5, 0.5, ...]
+ - Normal: return MinMaxScaler result as list
+ """
+ if not values:
+ return values
+ arr = np.asarray(values, dtype=float)
+ valid_mask = np.isfinite(arr)
+ if not valid_mask.any():
+ return values # all NaN → preserve
+ lo, hi = np.nanmin(arr), np.nanmax(arr)
+ if np.isclose(lo, hi):
+ return [0.5 if np.isfinite(v) else np.nan for v in values]
+ scaled = (arr - lo) / (hi - lo)
+ return scaled.tolist()
+
def calculate_pivot_metrics(
*,
previous_pos: int,
median_threshold_log = np.nanmedian(np.log1p(thresholds[start_pos:end_pos]))
amplitude_threshold_ratio = (
- amplitude / (amplitude + median_threshold_log)
+ amplitude / median_threshold_log
if np.isfinite(median_threshold_log) and median_threshold_log > 0
else np.nan
)
and np.isfinite(median_volume)
and median_volume > 0
):
- return avg_volume_per_candle / (avg_volume_per_candle + median_volume)
+ return avg_volume_per_candle / median_volume
return np.nan
def calculate_pivot_efficiency_ratio(
pivots_indices,
pivots_values_log,
pivots_directions,
- pivots_amplitudes,
- pivots_amplitude_threshold_ratios,
- pivots_volume_rates,
- pivots_speeds,
+ minmax_scale(pivots_amplitudes),
+ minmax_scale(pivots_amplitude_threshold_ratios),
+ minmax_scale(pivots_volume_rates),
+ minmax_scale(pivots_speeds),
pivots_efficiency_ratios,
pivots_volume_weighted_efficiency_ratios,
)