Main Analyzer

The main entry point for running trade geometry analysis.

analyzer

Main orchestrator for trade geometry analysis.

AnalysisConfig dataclass

Configuration for analysis run.

Source code in signal_analyzer/analyzer.py

@dataclass
class AnalysisConfig:
    """Configuration for analysis run."""

    H: int = 10
    """Forward horizon in bars"""

    sections: List[str] = field(default_factory=lambda: ["A", "B", "C", "D", "E", "F"])
    """Sections to run: A, B, C, D, E, F"""

    # Section A: Geometry
    trim_method: str | None = "iqr"
    trim_k: float = 1.5

    # Section B: Frontiers
    risk_q: float = 0.9
    opp_q: float = 0.8

    # Section D: TP/SL Feasibility
    tp_grid: np.ndarray | None = None
    sl_grid: np.ndarray | None = None
    store_paths: bool = True

    # Section E: Vol Normalization
    vol_col: str | None = None
    n_regimes: int = 3

    # Section F: Clusters
    n_clusters: int | None = None

H = 10 class-attribute instance-attribute

Forward horizon in bars

sections = field(default_factory=(lambda: ['A', 'B', 'C', 'D', 'E', 'F'])) class-attribute instance-attribute

Sections to run: A, B, C, D, E, F

AnalysisResult dataclass

Container for analysis results.

Source code in signal_analyzer/analyzer.py

@dataclass
class AnalysisResult:
    """Container for analysis results."""

    config: AnalysisConfig
    long_trades: TradeSet | None = None
    short_trades: TradeSet | None = None

    # Section results
    section_a: Dict[str, Any] = field(default_factory=dict)
    section_b: Dict[str, Any] = field(default_factory=dict)
    section_c: Dict[str, Any] = field(default_factory=dict)
    section_d: Dict[str, Any] = field(default_factory=dict)
    section_e: Dict[str, Any] = field(default_factory=dict)
    section_f: Dict[str, Any] = field(default_factory=dict)

    # Plots
    plots: Dict[str, Any] = field(default_factory=dict)

analyze(ohlc, sig_col='sig', config=None, open_col='Open', high_col='High', low_col='Low', sig_mode='transitions')

Run complete trade geometry analysis.

Parameters:

Name	Type	Description	Default
ohlc	DataFrame	OHLC data with signal column	required
sig_col	str	Signal column name	'sig'
config	AnalysisConfig	Analysis configuration. If None, uses defaults.	None
open_col	str	OHLC column names	'Open'
high_col	str	OHLC column names	'Open'
low_col	str	OHLC column names	'Open'
sig_mode	(transitions, levels)	Signal interpretation mode	'transitions'

Returns:

Type	Description
AnalysisResult	Container with all analysis results and plots

Notes

This is the main entry point for the library. It orchestrates all analysis sections based on the configuration.

Source code in signal_analyzer/analyzer.py

def analyze(
    ohlc: pd.DataFrame,
    sig_col: str = "sig",
    config: AnalysisConfig | None = None,
    open_col: str = "Open",
    high_col: str = "High",
    low_col: str = "Low",
    sig_mode: Literal["transitions", "levels"] = "transitions",
) -> AnalysisResult:
    """
    Run complete trade geometry analysis.

    Parameters
    ----------
    ohlc : pd.DataFrame
        OHLC data with signal column
    sig_col : str
        Signal column name
    config : AnalysisConfig, optional
        Analysis configuration. If None, uses defaults.
    open_col, high_col, low_col : str
        OHLC column names
    sig_mode : {'transitions', 'levels'}
        Signal interpretation mode

    Returns
    -------
    AnalysisResult
        Container with all analysis results and plots

    Notes
    -----
    This is the main entry point for the library. It orchestrates all
    analysis sections based on the configuration.
    """
    if config is None:
        config = AnalysisConfig()

    result = AnalysisResult(config=config)

    # 1. Extract entry events
    events = signal_to_events(ohlc, sig_col=sig_col, mode=sig_mode)

    # 2. Compute trade paths
    long_trades = None
    short_trades = None

    if len(events["enter_long"]) > 0:
        long_trades = compute_trade_paths(
            ohlc,
            events["enter_long"],
            config.H,
            side="long",
            open_col=open_col,
            high_col=high_col,
            low_col=low_col,
            vol_col=config.vol_col,
            store_paths=config.store_paths,
        )
        result.long_trades = long_trades

    if len(events["enter_short"]) > 0:
        short_trades = compute_trade_paths(
            ohlc,
            events["enter_short"],
            config.H,
            side="short",
            open_col=open_col,
            high_col=high_col,
            low_col=low_col,
            vol_col=config.vol_col,
            store_paths=config.store_paths,
        )
        result.short_trades = short_trades

    # 3. Run sections
    if "A" in config.sections:
        _run_section_a(result, config)

    if "B" in config.sections:
        _run_section_b(result, config)

    if "C" in config.sections:
        _run_section_c(result, config)

    if "D" in config.sections:
        _run_section_d(result, config)

    if "E" in config.sections and config.vol_col is not None:
        _run_section_e(result, config)

    if "F" in config.sections:
        _run_section_f(result, config)

    return result