Skip to content

Cluster Analysis

Section F: Trade archetype clustering and exit rule suggestions.

clusters

Clustering analysis: identify trade archetypes.

cluster_features(trades)

Construct feature matrix for clustering.

Parameters:

Name Type Description Default
trades TradeSet

Trade geometry data

 required

Returns:

Name Type Description
features ndarray

Feature matrix (n_trades x n_features)
feature_names list of str

Names of features
Notes

Features include: - MFE (max favorable excursion) - MAE (max adverse excursion) - MFE/|MAE| ratio (avoid div by zero) - t_mfe (time to reach MFE) - t_mae (time to reach MAE) - ordering_encoded (0=tie, 1=mfe_first, 2=mae_first)

Source code in signal_analyzer/analysis/clusters.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def cluster_features(trades: TradeSet) -> Tuple[np.ndarray, list[str]]:
    """
    Construct feature matrix for clustering.

    Parameters
    ----------
    trades : TradeSet
        Trade geometry data

    Returns
    -------
    features : np.ndarray
        Feature matrix (n_trades x n_features)
    feature_names : list of str
        Names of features

    Notes
    -----
    Features include:
    - MFE (max favorable excursion)
    - MAE (max adverse excursion)
    - MFE/|MAE| ratio (avoid div by zero)
    - t_mfe (time to reach MFE)
    - t_mae (time to reach MAE)
    - ordering_encoded (0=tie, 1=mfe_first, 2=mae_first)
    """
    mfe = trades.mfe
    mae = trades.mae
    dd = -mae  # Positive drawdown

    # Avoid division by zero in ratio
    ratio = np.zeros_like(mfe, dtype=float)
    valid_mask = dd > 0
    ratio[valid_mask] = mfe[valid_mask] / dd[valid_mask]

    # Ordering encoding
    labels = ordering_label(trades)
    ordering_encoded = np.zeros(len(labels))
    ordering_encoded[labels == "mfe_first"] = 1
    ordering_encoded[labels == "mae_first"] = 2

    # Time features
    t_mfe = trades.t_mfe.astype(float)
    t_mae = trades.t_mae.astype(float)

    features = np.column_stack([mfe, mae, ratio, t_mfe, t_mae, ordering_encoded])

    feature_names = [
        "mfe",
        "mae",
        "mfe_mae_ratio",
        "t_mfe",
        "t_mae",
        "ordering",
    ]

    return features, feature_names

cluster_trades(trades, n_clusters=None, max_clusters=6, method='kmeans', random_state=42)

Cluster trades into archetypes.

Parameters:

Name Type Description Default
trades TradeSet

Trade geometry data

 required
n_clusters int

Number of clusters. If None, automatically select using silhouette score.

 None
max_clusters int

Maximum number of clusters to test for auto-selection

 6
method (kmeans, hdbscan)

Clustering method

 'kmeans'
random_state int

Random seed for reproducibility

 42

Returns:

Type Description
dict

Dictionary containing: - 'labels': cluster labels (n_trades,) - 'n_clusters': number of clusters - 'features': feature matrix - 'feature_names': list of feature names - 'scaler': fitted StandardScaler - 'model': fitted clustering model
Source code in signal_analyzer/analysis/clusters.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def cluster_trades(
    trades: TradeSet,
    n_clusters: int | None = None,
    max_clusters: int = 6,
    method: str = "kmeans",
    random_state: int = 42,
) -> Dict[str, Any]:
    """
    Cluster trades into archetypes.

    Parameters
    ----------
    trades : TradeSet
        Trade geometry data
    n_clusters : int, optional
        Number of clusters. If None, automatically select using silhouette score.
    max_clusters : int
        Maximum number of clusters to test for auto-selection
    method : {'kmeans', 'hdbscan'}
        Clustering method
    random_state : int
        Random seed for reproducibility

    Returns
    -------
    dict
        Dictionary containing:
        - 'labels': cluster labels (n_trades,)
        - 'n_clusters': number of clusters
        - 'features': feature matrix
        - 'feature_names': list of feature names
        - 'scaler': fitted StandardScaler
        - 'model': fitted clustering model
    """
    features, feature_names = cluster_features(trades)

    # Standardize features
    scaler = StandardScaler()
    features_scaled = scaler.fit_transform(features)

    if method == "kmeans":
        if n_clusters is None:
            # Auto-select using silhouette score
            best_k = 3
            best_score = -1

            for k in range(2, max_clusters + 1):
                kmeans = KMeans(n_clusters=k, random_state=random_state, n_init=10)
                labels = kmeans.fit_predict(features_scaled)

                if len(np.unique(labels)) < 2:
                    continue

                score = silhouette_score(features_scaled, labels)

                if score > best_score:
                    best_score = score
                    best_k = k

            n_clusters = best_k

        model = KMeans(n_clusters=n_clusters, random_state=random_state, n_init=10)
        labels = model.fit_predict(features_scaled)

    elif method == "hdbscan":
        try:
            import hdbscan

            model = hdbscan.HDBSCAN(min_cluster_size=max(5, trades.n_trades // 20))
            labels = model.fit_predict(features_scaled)
            n_clusters = len(np.unique(labels[labels >= 0]))

        except ImportError:
            raise ImportError(
                "HDBSCAN not available. Install with: pip install hdbscan"
            )

    else:
        raise ValueError(f"Unknown method: {method}")

    return {
        "labels": labels,
        "n_clusters": n_clusters,
        "features": features,
        "features_scaled": features_scaled,
        "feature_names": feature_names,
        "scaler": scaler,
        "model": model,
    }

cluster_summary(trades, cluster_result)

Generate summary statistics for each cluster.

Parameters:

Name Type Description Default
trades TradeSet

Trade geometry data

 required
cluster_result dict

Output from cluster_trades

 required

Returns:

Type Description
dict

Dictionary mapping cluster_id -> summary dict with: - 'count': number of trades - 'median_mfe': median MFE - 'median_mae': median MAE - 'win_rate': fraction with MFE > 0 - 'mfe_first_rate': fraction where MFE came first - 'mae_first_rate': fraction where MAE came first - 'median_t_mfe': median time to MFE - 'median_t_mae': median time to MAE
Source code in signal_analyzer/analysis/clusters.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def cluster_summary(trades: TradeSet, cluster_result: Dict[str, Any]) -> Dict[int, Dict[str, Any]]:
    """
    Generate summary statistics for each cluster.

    Parameters
    ----------
    trades : TradeSet
        Trade geometry data
    cluster_result : dict
        Output from cluster_trades

    Returns
    -------
    dict
        Dictionary mapping cluster_id -> summary dict with:
        - 'count': number of trades
        - 'median_mfe': median MFE
        - 'median_mae': median MAE
        - 'win_rate': fraction with MFE > 0
        - 'mfe_first_rate': fraction where MFE came first
        - 'mae_first_rate': fraction where MAE came first
        - 'median_t_mfe': median time to MFE
        - 'median_t_mae': median time to MAE
    """
    labels = cluster_result["labels"]
    cluster_ids = np.unique(labels)

    # Filter out noise points (label = -1 for HDBSCAN)
    cluster_ids = cluster_ids[cluster_ids >= 0]

    summaries = {}

    for cluster_id in cluster_ids:
        mask = labels == cluster_id

        cluster_mfe = trades.mfe[mask]
        cluster_mae = trades.mae[mask]
        cluster_t_mfe = trades.t_mfe[mask]
        cluster_t_mae = trades.t_mae[mask]

        # Ordering
        ordering = ordering_label(trades)
        cluster_ordering = ordering[mask]

        summaries[int(cluster_id)] = {
            "count": int(np.sum(mask)),
            "median_mfe": float(np.median(cluster_mfe)),
            "median_mae": float(np.median(cluster_mae)),
            "win_rate": float(np.mean(cluster_mfe > 0)),
            "mfe_first_rate": float(np.mean(cluster_ordering == "mfe_first")),
            "mae_first_rate": float(np.mean(cluster_ordering == "mae_first")),
            "median_t_mfe": float(np.median(cluster_t_mfe)),
            "median_t_mae": float(np.median(cluster_t_mae)),
        }

    return summaries

suggest_exit_rules(cluster_summaries, mfe_threshold_fast=1.0, dd_threshold_tight=1.0)

Suggest exit rule archetypes based on cluster characteristics.

Parameters:

Name Type Description Default
cluster_summaries dict

Output from cluster_summary

 required
mfe_threshold_fast float

MFE threshold to consider "fast winner"

 1.0
dd_threshold_tight float

DD threshold to consider "tight stop friendly"

 1.0

Returns:

Type Description
dict

Dictionary mapping cluster_id -> archetype description
Source code in signal_analyzer/analysis/clusters.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def suggest_exit_rules(
    cluster_summaries: Dict[int, Dict[str, Any]],
    mfe_threshold_fast: float = 1.0,
    dd_threshold_tight: float = 1.0,
) -> Dict[int, str]:
    """
    Suggest exit rule archetypes based on cluster characteristics.

    Parameters
    ----------
    cluster_summaries : dict
        Output from cluster_summary
    mfe_threshold_fast : float
        MFE threshold to consider "fast winner"
    dd_threshold_tight : float
        DD threshold to consider "tight stop friendly"

    Returns
    -------
    dict
        Dictionary mapping cluster_id -> archetype description
    """
    archetypes = {}

    for cluster_id, summary in cluster_summaries.items():
        mfe = summary["median_mfe"]
        dd = -summary["median_mae"]
        mfe_first_rate = summary["mfe_first_rate"]
        win_rate = summary["win_rate"]

        # Classify archetype
        if mfe > mfe_threshold_fast and dd < dd_threshold_tight and mfe_first_rate > 0.6:
            archetype = "Fast Winner (trailing stop)"
        elif mfe > mfe_threshold_fast and dd > dd_threshold_tight and mfe_first_rate < 0.4:
            archetype = "Needs Room (wide SL, time stop)"
        elif mfe < 0.5 and win_rate < 0.4:
            archetype = "Noise (consider filtering)"
        elif mfe > 2.0 and win_rate > 0.6:
            archetype = "Strong Edge (hold longer)"
        else:
            archetype = "Mixed (context-dependent)"

        archetypes[cluster_id] = archetype

    return archetypes