diff --git a/src/agent_loop/confidence_calibrator.py b/src/agent_loop/confidence_calibrator.py index 8b623e7..d7b2d60 100644 --- a/src/agent_loop/confidence_calibrator.py +++ b/src/agent_loop/confidence_calibrator.py @@ -58,8 +58,10 @@ def calibrate( Adjustments applied in order: 1. Historical accuracy-based adjustment (per-action) - 2. Per-sector accuracy adjustment - 3. Regime-based adjustment + 2. Regime-based adjustment + + ``sector`` is accepted but not yet used — per-sector calibration needs + sector metadata on the decisions table, which does not exist yet. """ adjustment = 0.0 @@ -67,12 +69,7 @@ def calibrate( action_adj = self._action_calibration(action) adjustment += action_adj - # 2. Sector-level calibration - if sector: - sector_adj = self._sector_calibration(sector) - adjustment += sector_adj - - # 3. Regime adjustment + # 2. Regime adjustment regime_adj = self._regime_adjustment(regime, action) adjustment += regime_adj @@ -83,13 +80,12 @@ def calibrate( if abs(adjustment) > 0.01: logger.debug( - "Calibration %s %s: %.2f → %.2f (action=%.3f sector=%.3f regime=%.3f)", + "Calibration %s %s: %.2f → %.2f (action=%.3f regime=%.3f)", action, symbol, raw_confidence, calibrated, action_adj, - sector_adj if sector else 0.0, regime_adj, ) @@ -263,12 +259,6 @@ def _action_calibration(self, action: str) -> float: finally: conn.close() - def _sector_calibration(self, sector: str) -> float: - """Compute confidence adjustment based on sector-level accuracy.""" - # Sector tracking would require joining decisions with thesis/symbol metadata. - # For now, return 0 — will be enhanced when we add sector to decisions table. - return 0.0 - def _regime_adjustment(self, regime: str, action: str) -> float: """Apply regime-based confidence adjustment.""" adjustments = self._regime_adjustments.get(regime, {}) diff --git a/src/prediction/analysis_frameworks.py b/src/prediction/analysis_frameworks.py index 300f9ab..8f0ca86 100644 --- a/src/prediction/analysis_frameworks.py +++ b/src/prediction/analysis_frameworks.py @@ -15,56 +15,6 @@ import math from typing import Any -# --------------------------------------------------------------------------- -# Legacy frameworks — DEPRECATED -# Use SEVEN_DIMENSION_FRAMEWORK (full analysis) or QUICK_DIMENSION_FRAMEWORK -# (quick insights) instead. Kept only for backward compatibility. -# --------------------------------------------------------------------------- - -PROFESSIONAL_ANALYSIS_FRAMEWORK = """\ -You are a professional A-share analyst combining multi-school investment methodologies. \ -Follow this framework strictly when analyzing. -Write all output text in Chinese. - -## Analysis Methodology - -### 1. Quantitative Factor Analysis (AQR / Two Sigma style) -- **Momentum factor**: Recent trend direction, MA alignment (bullish 多头 / bearish 空头 / converging 粘合) -- **Volatility factor**: Recent amplitude changes, Bollinger Band width -- **Volume factor**: Volume ratio changes, price-volume correlation - -### 2. Value Investing Check (Buffett framework) -- **Margin of safety**: Is the current valuation level reasonable? -- **Trend confirmation**: Does the medium-to-long term trend support the current judgment? - -### 3. Contrarian Thinking Check (Munger framework) -- **Inversion analysis**: What scenarios would invalidate the current judgment? -- **Psychological bias check**: Recency bias (recent moves distorting objectivity), anchoring (fixation on historical highs/lows), herding (is the sector overheated?) -- **Multi-factor cross-validation**: Are technicals, capital flow, and news aligned? - -### 4. A-share Specific Dimensions -- **Policy sensitivity**: Impact of policy direction on the industry/stock -- **Capital flow**: Direction and magnitude of institutional money flow (super-large + large order net inflow) -- **Sector linkage**: Overall sector performance, concept rotation position -- **Price limit mechanism**: Consider the stock's daily price limit when analyzing - -## Data Quality Rules -- If data is labeled "non-realtime" or "historical", state this clearly in the analysis -- Do not misattribute other sectors' market conditions to the current stock -- Quantitative strategy signals serve only as one reference dimension — never let them alone determine the conclusion -- Bayesian probabilities provide historical statistical support, but consider whether the current market environment is comparable to historical patterns -""" - -# DEPRECATED — use QUICK_DIMENSION_FRAMEWORK instead -QUICK_ANALYSIS_FRAMEWORK = """\ -You are a professional A-share analyst. Key analysis points: -Write all output text in Chinese. -1. Synthesize quantitative signals (strategy consensus, Bayesian probability) with technicals to form a judgment -2. Account for the stock's daily price limit and sector classification -3. If data is labeled non-realtime, state this clearly -4. Consider reversal risk — never blindly follow a single signal -""" - # ═══════════════════════════════════════════════════════════════════════════ # v7.0 Seven-Dimension Framework (FR-PR001) # ═══════════════════════════════════════════════════════════════════════════ diff --git a/src/quant/alternative_bars.py b/src/quant/alternative_bars.py deleted file mode 100644 index c04b2fb..0000000 --- a/src/quant/alternative_bars.py +++ /dev/null @@ -1,382 +0,0 @@ -"""Alternative bar generator for advanced signal detection. - -Time bars oversample quiet periods and undersample volatile ones. -Alternative bar types respond to actual market activity: -- Volume bars: new bar every N shares traded -- Dollar/Amount bars: new bar every N RMB traded -- Tick imbalance bars: new bar when buy/sell imbalance exceeds threshold - -Reference: Lopez de Prado, "Advances in Financial Machine Learning", Ch. 2 -""" - -from __future__ import annotations - -import numpy as np -import pandas as pd -from dataclasses import dataclass - -from src.utils.logger import get_logger - -logger = get_logger("quant.alternative_bars") - -__all__ = [ - "AlternativeBarGenerator", - "BarConfig", -] - -_EMPTY_BAR_COLUMNS = [ - "datetime", - "open", - "high", - "low", - "close", - "volume", - "amount", - "bar_count", -] - - -@dataclass -class BarConfig: - """Configuration for alternative bar generation.""" - - bar_type: str # "volume", "amount", "tick_imbalance" - threshold: float # Volume, amount, or imbalance threshold - - -class AlternativeBarGenerator: - """Generate alternative bar types from tick or minute data.""" - - def volume_bars(self, data: pd.DataFrame, threshold: int = 100_000) -> pd.DataFrame: - """Generate volume bars -- new bar every ``threshold`` shares. - - Args: - data: DataFrame with columns - [datetime, open/price, high, low, close, volume, amount]. - Can be tick-level or minute-level data. - threshold: Volume per bar (shares). - - Returns: - DataFrame with columns - [datetime, open, high, low, close, volume, amount, bar_count] - where each row represents one volume bar. - """ - return self._accumulation_bars(data, threshold, accumulate_col="volume") - - def amount_bars( - self, data: pd.DataFrame, threshold: float = 5_000_000 - ) -> pd.DataFrame: - """Generate dollar/amount bars -- new bar every ``threshold`` RMB traded. - - Args: - data: DataFrame with volume and amount columns. - threshold: Amount per bar (RMB). Default 500万. - """ - return self._accumulation_bars(data, threshold, accumulate_col="amount") - - def tick_imbalance_bars( - self, - ticks: list, - expected_imbalance: float | None = None, - ) -> pd.DataFrame: - """Generate tick imbalance bars. - - A tick is classified as buy (+1) or sell (-1). - Cumulative imbalance = sum of classifications. - New bar when |cumulative imbalance| >= threshold. - - If ``expected_imbalance`` is None, use exponential moving average - of previous bar imbalances as adaptive threshold (per Lopez de Prado). - - Args: - ticks: List of objects with attributes: - datetime, price, volume, amount, direction (+1/-1). - expected_imbalance: Fixed threshold, or None for adaptive. - """ - if not ticks: - return self._empty_bars() - - # Extract tick data - tick_data: list[dict] = [] - for t in ticks: - tick_data.append( - { - "datetime": getattr(t, "datetime", None), - "price": getattr(t, "price", 0.0), - "volume": getattr(t, "volume", 0), - "amount": getattr(t, "amount", 0.0), - "direction": getattr(t, "direction", 0), - } - ) - - if not tick_data: - return self._empty_bars() - - # Adaptive threshold: start with a default then update via EWMA - ewma_alpha = 0.1 - threshold = expected_imbalance if expected_imbalance is not None else 20.0 - adaptive = expected_imbalance is None - - bars: list[dict] = [] - bar_open = tick_data[0]["price"] - bar_high = tick_data[0]["price"] - bar_low = tick_data[0]["price"] - bar_start_dt = tick_data[0]["datetime"] - bar_volume = 0 - bar_amount = 0.0 - cumulative_imbalance = 0 - bar_tick_count = 0 - - for tick in tick_data: - price = tick["price"] - direction = tick["direction"] - - if price <= 0: - continue - - bar_high = max(bar_high, price) - bar_low = min(bar_low, price) - bar_close = price - bar_volume += tick["volume"] - bar_amount += tick["amount"] - bar_tick_count += 1 - - # Classify tick direction - if direction > 0: - cumulative_imbalance += 1 - elif direction < 0: - cumulative_imbalance -= 1 - - # Check if we should close this bar - if abs(cumulative_imbalance) >= threshold and bar_tick_count > 0: - bars.append( - { - "datetime": bar_start_dt, - "open": bar_open, - "high": bar_high, - "low": bar_low, - "close": bar_close, - "volume": bar_volume, - "amount": bar_amount, - "bar_count": len(bars) + 1, - } - ) - - # Update adaptive threshold via EWMA - if adaptive: - threshold = ( - ewma_alpha * abs(cumulative_imbalance) - + (1 - ewma_alpha) * threshold - ) - # Floor to avoid collapsing to zero - threshold = max(threshold, 2.0) - - # Reset for next bar - cumulative_imbalance = 0 - bar_tick_count = 0 - bar_volume = 0 - bar_amount = 0.0 - # Next tick sets new bar_open - bar_open = price - bar_high = price - bar_low = price - bar_start_dt = tick["datetime"] - - # Don't emit partial bar (consistent with Lopez de Prado) - - if not bars: - return self._empty_bars() - - return pd.DataFrame(bars, columns=_EMPTY_BAR_COLUMNS) - - def compute_factors_from_bars(self, bars: pd.DataFrame) -> dict[str, float]: - """Compute enhanced factors from alternative bars. - - These factors capture information lost in time bars: - - bar_frequency: bars per unit time (high = volatile) - - bar_size_consistency: std(volume) / mean(volume) of bars - - recent_bar_direction: bullish/bearish ratio of last N bars - - bar_acceleration: bar frequency change (speeding up vs slowing) - """ - if bars is None or bars.empty or len(bars) < 2: - return { - "bar_frequency": 0.5, - "bar_size_consistency": 0.5, - "recent_bar_direction": 0.5, - "bar_acceleration": 0.5, - } - - factors: dict[str, float] = {} - - # --- Bar frequency: bars per minute --- - if "datetime" in bars.columns and len(bars) >= 2: - dt_col = bars["datetime"] - if not pd.api.types.is_datetime64_any_dtype(dt_col): - dt_col = pd.to_datetime(dt_col, errors="coerce") - - valid_dt = dt_col.dropna() - if len(valid_dt) >= 2: - span_seconds = (valid_dt.iloc[-1] - valid_dt.iloc[0]).total_seconds() - if span_seconds > 0: - bars_per_min = len(bars) / (span_seconds / 60.0) - # Normalize: 0 bars/min → 0, 2+ bars/min → 1 - factors["bar_frequency"] = round( - max(0.0, min(1.0, bars_per_min / 2.0)), 4 - ) - - factors.setdefault("bar_frequency", 0.5) - - # --- Size consistency: low CV = institutional-like flow --- - if "volume" in bars.columns: - vol = bars["volume"].astype(float) - mean_vol = vol.mean() - if mean_vol > 0: - cv = vol.std() / mean_vol - # Low CV = consistent = higher score - factors["bar_size_consistency"] = round(max(0.0, min(1.0, 1.0 - cv)), 4) - else: - factors["bar_size_consistency"] = 0.5 - else: - factors["bar_size_consistency"] = 0.5 - - # --- Recent bar direction --- - if "close" in bars.columns and "open" in bars.columns: - recent = bars.tail(min(10, len(bars))) - bullish = (recent["close"] > recent["open"]).sum() - total = len(recent) - factors["recent_bar_direction"] = round(bullish / total, 4) - else: - factors["recent_bar_direction"] = 0.5 - - # --- Acceleration: compare bar frequency in recent vs earlier half --- - if len(bars) >= 4 and "datetime" in bars.columns: - dt_col = bars["datetime"] - if not pd.api.types.is_datetime64_any_dtype(dt_col): - dt_col = pd.to_datetime(dt_col, errors="coerce") - - valid_dt = dt_col.dropna() - if len(valid_dt) >= 4: - mid = len(valid_dt) // 2 - first_half_span = ( - valid_dt.iloc[mid] - valid_dt.iloc[0] - ).total_seconds() - second_half_span = ( - valid_dt.iloc[-1] - valid_dt.iloc[mid] - ).total_seconds() - - if first_half_span > 0 and second_half_span > 0: - first_rate = mid / first_half_span - second_rate = (len(valid_dt) - mid) / second_half_span - # Ratio > 1 = accelerating (more bars recently) - accel_ratio = second_rate / first_rate - # Sigmoid: ratio 1.0 → 0.5, >1 → higher, <1 → lower - import math - - factors["bar_acceleration"] = round( - 1.0 / (1.0 + math.exp(-(accel_ratio - 1.0) * 3.0)), 4 - ) - - factors.setdefault("bar_acceleration", 0.5) - - return factors - - # ------------------------------------------------------------------ - # Internal helpers - # ------------------------------------------------------------------ - - def _accumulation_bars( - self, - data: pd.DataFrame, - threshold: float, - accumulate_col: str, - ) -> pd.DataFrame: - """Generic accumulation bar builder (shared by volume and amount bars). - - Iterates through source rows, accumulating ``accumulate_col`` until - the cumulative value reaches ``threshold``, then emits a bar. - """ - if data is None or data.empty: - return self._empty_bars() - - if threshold <= 0: - logger.warning("Bar threshold must be positive, got %s", threshold) - return self._empty_bars() - - df = data.copy() - - # Normalize column names: accept 'price' as 'close' for tick data - if "close" not in df.columns and "price" in df.columns: - df["close"] = df["price"] - if "open" not in df.columns: - df["open"] = df["close"] - if "high" not in df.columns: - df["high"] = df["close"] - if "low" not in df.columns: - df["low"] = df["close"] - if "amount" not in df.columns: - df["amount"] = 0.0 - if "volume" not in df.columns: - df["volume"] = 0 - - required = ["datetime", "open", "high", "low", "close", accumulate_col] - for col in required: - if col not in df.columns: - logger.warning("Missing column %s for bar generation", col) - return self._empty_bars() - - # Convert to numpy for faster iteration - dt_vals = df["datetime"].values - open_vals = df["open"].values.astype(float) - high_vals = df["high"].values.astype(float) - low_vals = df["low"].values.astype(float) - close_vals = df["close"].values.astype(float) - vol_vals = df["volume"].values.astype(float) - amt_vals = df["amount"].values.astype(float) - accum_vals = df[accumulate_col].values.astype(float) - - bars: list[dict] = [] - bar_start_idx = 0 - cumulative = 0.0 - - for i in range(len(df)): - cumulative += accum_vals[i] - - if cumulative >= threshold: - # Emit bar from bar_start_idx to i (inclusive) - bar_open = open_vals[bar_start_idx] - bar_high = float(np.max(high_vals[bar_start_idx : i + 1])) - bar_low = float(np.min(low_vals[bar_start_idx : i + 1])) - bar_close = close_vals[i] - bar_vol = float(np.sum(vol_vals[bar_start_idx : i + 1])) - bar_amt = float(np.sum(amt_vals[bar_start_idx : i + 1])) - - bars.append( - { - "datetime": dt_vals[bar_start_idx], - "open": bar_open, - "high": bar_high, - "low": bar_low, - "close": bar_close, - "volume": bar_vol, - "amount": bar_amt, - "bar_count": len(bars) + 1, - } - ) - - # Handle overflow: if cumulative > threshold, the remainder - # carries into the next bar - overflow = cumulative - threshold - cumulative = overflow - bar_start_idx = i + 1 - - # Don't emit partial bar at the end (consistent with academic practice) - - if not bars: - return self._empty_bars() - - return pd.DataFrame(bars, columns=_EMPTY_BAR_COLUMNS) - - @staticmethod - def _empty_bars() -> pd.DataFrame: - """Return empty DataFrame with correct bar columns.""" - return pd.DataFrame(columns=_EMPTY_BAR_COLUMNS) diff --git a/src/recommendation/overnight_risk.py b/src/recommendation/overnight_risk.py index 948f562..a5c4dc3 100644 --- a/src/recommendation/overnight_risk.py +++ b/src/recommendation/overnight_risk.py @@ -37,25 +37,6 @@ class OvernightRiskProfile: # Risk score (0=safe, 1=dangerous) risk_score: float - def to_context_str(self) -> str: - """Format as context string for LLM prompt injection.""" - parts = [ - f"隔夜风险分析 ({self.symbol}):", - f" 平均隔夜跳空: {self.avg_gap_pct:+.2f}% (标准差: {self.std_gap_pct:.2f}%)", - f" 最大负跳空: {self.max_negative_gap_pct:.2f}%", - f" 隔夜跳空为负概率: {self.gap_down_ratio:.0%}", - ] - if self.rally_sample_size > 0: - parts.extend( - [ - f" 大涨后次日回调概率: {self.post_rally_drawdown_prob:.0%} " - f"(样本={self.rally_sample_size}天)", - f" 大涨后次日平均收益: {self.post_rally_avg_return:+.2f}%", - ] - ) - parts.append(f" 综合隔夜风险评分: {self.risk_score:.2f}/1.00") - return "\n".join(parts) - class OvernightRiskCalculator: """Calculate overnight risk metrics from historical OHLCV data.""" diff --git a/src/recommendation/rec_store.py b/src/recommendation/rec_store.py index 5bfbf8a..95213ea 100644 --- a/src/recommendation/rec_store.py +++ b/src/recommendation/rec_store.py @@ -693,38 +693,6 @@ def get_style_win_rates(self, days: int = 30) -> dict[str, dict[str, Any]]: finally: conn.close() - def get_sector_win_rates(self, days: int = 30) -> dict[str, dict[str, Any]]: - """Get per-sector win rates for intel chain feedback. - - Returns dict like: {"银行": {"win_rate_t1": 0.7, "count": 10}} - """ - conn = self._connect() - try: - cutoff = (datetime.now(UTC) - timedelta(days=days)).strftime("%Y-%m-%d") - rows = conn.execute( - """ - SELECT json_extract(r.factors, '$.sector_momentum') as sector_score, - r.style, - COUNT(*) as count, - SUM(CASE WHEN o.correct_t1 = 1 THEN 1 ELSE 0 END) as wins_t1, - AVG(o.actual_change_t1) as avg_return_t1 - FROM recommendations r - JOIN recommendation_outcomes o ON r.id = o.rec_id - WHERE r.created_at >= ? - AND o.actual_change_t1 IS NOT NULL - GROUP BY r.style - HAVING count >= 3 - """, - (cutoff,), - ).fetchall() - - return {dict(r)["style"]: dict(r) for r in rows} - except Exception as exc: - logger.warning("Failed to get sector win rates: %s", exc) - return {} - finally: - conn.close() - # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ diff --git a/src/web/services/broker_interface.py b/src/web/services/broker_interface.py index 509ff3d..2568cba 100644 --- a/src/web/services/broker_interface.py +++ b/src/web/services/broker_interface.py @@ -240,27 +240,28 @@ def _read_portfolio() -> dict[str, Any]: class LiveBroker(BrokerInterface): - """Stub for real broker integration. + """Intentionally-unimplemented stub for real broker integration. - Requires: - - config/broker.yaml: mode=live - - Explicit user configuration - - All trades must pass through ConfirmationGate with VERIFIED status - - NOT IMPLEMENTED — placeholder for future integration. + This is a **simulation-only** project — live order routing is not implemented + by design. Every method raises :class:`NotImplementedError`; use + :class:`SimulationBroker` (the default) instead. A real integration would + require ``config/broker.yaml: mode=live``, explicit user configuration, and + all trades passing through ``ConfirmationGate`` with VERIFIED status. """ + _MSG = "LiveBroker is not implemented — this is a simulation-only project; use SimulationBroker." + def __init__(self) -> None: - logger.warning("LiveBroker instantiated — NOT IMPLEMENTED") + logger.warning("LiveBroker instantiated — not implemented (simulation-only)") def get_positions(self) -> list[Position]: - raise NotImplementedError("LiveBroker not implemented") + raise NotImplementedError(self._MSG) def get_balance(self) -> Balance: - raise NotImplementedError("LiveBroker not implemented") + raise NotImplementedError(self._MSG) def get_order_status(self, order_id: str) -> OrderStatus: - raise NotImplementedError("LiveBroker not implemented") + raise NotImplementedError(self._MSG) def submit_order( self, @@ -270,7 +271,7 @@ def submit_order( price: float, gate_request_id: str = "", ) -> OrderStatus: - raise NotImplementedError("LiveBroker not implemented") + raise NotImplementedError(self._MSG) @property def mode(self) -> str: diff --git a/tests/unit/test_alternative_bars.py b/tests/unit/test_alternative_bars.py deleted file mode 100644 index a402fb3..0000000 --- a/tests/unit/test_alternative_bars.py +++ /dev/null @@ -1,151 +0,0 @@ -"""Tests for AlternativeBarGenerator -- volume, amount, and imbalance bars.""" - -from __future__ import annotations - -import numpy as np -import pandas as pd -import pytest -from datetime import datetime, timedelta - - -def _make_minute_data(n=100, base_price=10.0): - np.random.seed(42) - dates = [datetime(2026, 3, 10, 9, 30) + timedelta(minutes=i) for i in range(n)] - return pd.DataFrame( - { - "datetime": dates, - "open": [base_price + np.random.normal(0, 0.05) for _ in range(n)], - "high": [base_price + 0.1 + np.random.normal(0, 0.05) for _ in range(n)], - "low": [base_price - 0.1 + np.random.normal(0, 0.05) for _ in range(n)], - "close": [base_price + np.random.normal(0, 0.05) for _ in range(n)], - "volume": [int(50000 + np.random.normal(0, 10000)) for _ in range(n)], - "amount": [ - base_price * 50000 + np.random.normal(0, 50000) for _ in range(n) - ], - } - ) - - -class TestAlternativeBarGenerator: - @pytest.fixture() - def generator(self): - from src.quant.alternative_bars import AlternativeBarGenerator - - return AlternativeBarGenerator() - - def test_volume_bars_returns_df(self, generator): - data = _make_minute_data() - result = generator.volume_bars(data, threshold=200000) - assert isinstance(result, pd.DataFrame) - assert len(result) > 0 - - def test_volume_bars_correct_volume(self, generator): - """Each volume bar should have roughly threshold volume (overflow may reduce next bar).""" - data = _make_minute_data() - threshold = 200000 - result = generator.volume_bars(data, threshold=threshold) - if len(result) > 1: - # Total volume across all bars should be roughly correct - total_vol = result["volume"].sum() - assert total_vol > 0 - # Average bar volume should be near the threshold - avg_vol = total_vol / len(result) - assert avg_vol >= threshold * 0.5 - - def test_amount_bars_returns_df(self, generator): - data = _make_minute_data() - result = generator.amount_bars(data, threshold=2_000_000) - assert isinstance(result, pd.DataFrame) - - def test_empty_data(self, generator): - result = generator.volume_bars(pd.DataFrame(), threshold=100000) - assert isinstance(result, pd.DataFrame) - assert result.empty - - def test_none_data(self, generator): - result = generator.volume_bars(None, threshold=100000) - assert isinstance(result, pd.DataFrame) - assert result.empty - - def test_compute_factors(self, generator): - data = _make_minute_data() - bars = generator.volume_bars(data, threshold=200000) - factors = generator.compute_factors_from_bars(bars) - assert isinstance(factors, dict) - for key in [ - "bar_frequency", - "bar_size_consistency", - "recent_bar_direction", - "bar_acceleration", - ]: - assert key in factors - assert 0.0 <= factors[key] <= 1.0 - - def test_compute_factors_empty_bars(self, generator): - factors = generator.compute_factors_from_bars(pd.DataFrame()) - assert all(v == 0.5 for v in factors.values()) - - def test_compute_factors_none(self, generator): - factors = generator.compute_factors_from_bars(None) - assert all(v == 0.5 for v in factors.values()) - - def test_tick_imbalance_bars(self, generator): - """Tick imbalance bars expect numeric direction (+1/-1).""" - from dataclasses import dataclass - - @dataclass - class NumericTick: - datetime: object - price: float - volume: int - amount: float - direction: int - - np.random.seed(42) - ticks = [] - for i in range(100): - d = 1 if i % 3 != 0 else -1 - ticks.append( - NumericTick( - datetime=datetime(2026, 3, 10, 9, 30, i % 60), - price=10.0 + np.random.normal(0, 0.01), - volume=100, - amount=1000, - direction=d, - ) - ) - result = generator.tick_imbalance_bars(ticks, expected_imbalance=10.0) - assert isinstance(result, pd.DataFrame) - - def test_tick_imbalance_bars_empty(self, generator): - result = generator.tick_imbalance_bars([], expected_imbalance=10.0) - assert isinstance(result, pd.DataFrame) - assert result.empty - - def test_volume_bars_columns(self, generator): - data = _make_minute_data() - result = generator.volume_bars(data, threshold=200000) - if not result.empty: - expected_cols = [ - "datetime", - "open", - "high", - "low", - "close", - "volume", - "amount", - "bar_count", - ] - for col in expected_cols: - assert col in result.columns, f"Missing column: {col}" - - def test_amount_bars_threshold(self, generator): - """Amount bars should accumulate until threshold RMB is reached.""" - data = _make_minute_data() - threshold = 1_000_000 - result = generator.amount_bars(data, threshold=threshold) - assert isinstance(result, pd.DataFrame) - if len(result) > 1: - # Average bar amount should be near the threshold - avg_amt = result["amount"].mean() - assert avg_amt >= threshold * 0.5 diff --git a/tests/unit/test_overnight_risk.py b/tests/unit/test_overnight_risk.py index a97d552..6a5e365 100644 --- a/tests/unit/test_overnight_risk.py +++ b/tests/unit/test_overnight_risk.py @@ -102,17 +102,6 @@ def test_rally_detection(self): assert profile.rally_sample_size > 0 assert isinstance(profile.post_rally_drawdown_prob, float) - def test_context_str(self): - calc = OvernightRiskCalculator() - closes = [100 + i for i in range(15)] - opens = [99 + i for i in range(15)] - df = self._make_df(closes, opens) - profile = calc._compute_profile("000001", df, rally_threshold=5.0) - - ctx = profile.to_context_str() - assert "隔夜风险分析" in ctx - assert "000001" in ctx - def test_batch_calculation(self): # Mock fetcher to return empty — should return empty dict gracefully fetcher = Mock()