Skip to content

artifact_cleanup

Artifact Cleanup Utilities

Functions for removing invalid segments and filling in gaps with flat segments.

clean_artifacts 

clean_artifacts(df, value_col, segments_refined, method_params, inverse_only=False)

Removes segments any invalid segments, such as inversions or overlaps. Typically to clean up after boundary adjustments introduced from noise or trend refinements.

Parameters:

  • segments_refined

    (list) –

    List of segment dictionaries potentially with artifacts from post-processing.

  • method_params

    (dict) –

    Optional parameters for cleanup behavior. Supported keys:

    • is_abrupt_padded (bool): If True, skips neighboring-noise checks around abrupt segments. Defaults to False.
    • abrupt_padding (int): Padding window in days used by abrupt refinement; included for pipeline consistency. Defaults to 28.

  • inverse_only

    (bool, default: False ) –

    If True, only perform inverse checks and skip other artifact cleanups. Useful for final cleanup pass after flat fill ins.

Returns:

  • list ( list[dict] ) –

    Cleaned segment list with only valid-length segments.

Source code in pytrendy/post_processing/segments_refine/artifact_cleanup.py 
 12
 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
 70
 71
 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
161
162
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
219
220
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
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def clean_artifacts(df: pd.DataFrame, value_col: str, segments_refined: list[dict], method_params: dict, inverse_only: bool = False) -> list[dict]:
    """
    Removes segments any invalid segments, such as inversions or overlaps.
    Typically to clean up after boundary adjustments introduced from noise or trend refinements.

    Args:
        segments_refined (list): List of segment dictionaries potentially with artifacts from post-processing.
        method_params (dict): Optional parameters for cleanup behavior. Supported keys:

            - **is_abrupt_padded** (`bool`): If `True`, skips neighboring-noise checks around abrupt segments. Defaults to `False`.
            - **abrupt_padding** (`int`): Padding window in days used by abrupt refinement; included for pipeline consistency. Defaults to `28`.
        inverse_only (bool): If True, only perform inverse checks and skip other artifact cleanups. Useful for final cleanup pass after flat fill ins.

    Returns:
        list: Cleaned segment list with only valid-length segments.
    """

    def has_inverse(df: pd.DataFrame, value_col: str, segment: dict) -> bool:
        """
        Checks that if end moved before start from neighbour adjustment, removes artifact.
        Also if trend, but total_change is actually in opposing direction, also remove
        """
        start = pd.to_datetime(segment['start'])
        end =  pd.to_datetime(segment['end'])
        is_flat = segment['direction'] == 'Flat'
        is_border = (start == df.index[0]) or (end == df.index[-1])
        flat_edge_case = is_flat and not is_border

        # inverse if start before end, immediately clean
        if (end - start).days < 0:
            return True

        # if length 0, but not from flat fill in middle, then clean
        if (end - start).days == 0 and not flat_edge_case: 
            return True

        # inverse if tagged direction does not match total change
        total_change = df.loc[start:end, value_col].diff().sum()
        if \
            (segment['direction'] == 'Up' and total_change <= 0) or \
            (segment['direction'] == 'Down' and total_change >= 0):
            return True

        return False

    def has_overlap_next(segment: dict, segment_next: dict) -> bool:
        """Checks whether overlap exists between curr & next, and current is more insignificant"""
        dir = segment['direction']
        start =  pd.to_datetime(segment['start'])
        end =  pd.to_datetime(segment['end'])
        width = (end - start).days

        next_dir = segment_next['direction']
        next_start = pd.to_datetime(segment_next['start'])
        next_end = pd.to_datetime(segment_next['end'])
        next_width = (next_end - next_start).days

        # Define conditions # TODO: Cleanup redunant condition statements no longer used.
        is_overlap_next = (end >= next_start)
        is_same_dir = (dir == next_dir)
        is_curr_shorter = (width <= next_width)
        is_curr_similar = (next_width <= 1.5 * width) and (next_width >= 0.5 * width)

        is_trend = (dir in ('Up', 'Down'))
        is_next_noise = (next_dir == 'Noise')
        is_next_opposite_trend = (next_dir in ('Up', 'Down') and next_dir != dir)
        is_next_flat = (next_dir == 'Flat')

        is_next_gradual = ('trend_class' in segment_next and segment_next['trend_class'] == 'gradual')
        is_next_abrupt = ('trend_class' in segment_next and segment_next['trend_class'] == 'abrupt')

        # Trigger edge cases of overlap if satisfied
        if is_overlap_next and is_same_dir:
            return True # overlap when same direction, and is same dir

        return False

    def has_overlap_prev(segment: dict, segment_prev: dict) -> bool:
        """Light checks with overlaps on previous, that wouldnt already be covered by has_overlap_next"""
        dir = segment['direction']
        start =  pd.to_datetime(segment['start'])
        end =  pd.to_datetime(segment['end'])
        width = (end - start).days

        prev_dir = segment_prev['direction']
        prev_start = pd.to_datetime(segment_prev['start'])
        prev_end = pd.to_datetime(segment_prev['end'])
        prev_width = (prev_end - prev_start).days

        # Define conditions # TODO: Cleanup redunant condition statements no longer used.
        is_overlap_prev = (start <= prev_end)
        is_curr_shorter = (width <= prev_width)
        is_curr_similar = (prev_width <= 1.5 * width) and (prev_width >= 0.5 * width)

        is_trend = (dir in ('Up', 'Down'))
        is_prev_noise = (prev_dir == 'Noise')
        is_prev_opposite_trend = (prev_dir in ('Up', 'Down') and prev_dir != dir)
        is_prev_flat = (prev_dir == 'Flat')

        if is_overlap_prev and (is_trend and (is_prev_noise or is_prev_opposite_trend) and is_curr_shorter):
            return True # overlap when curr is trend and prev is noise of larger/equal window
        if is_overlap_prev and (is_trend and is_prev_flat) and is_curr_similar:
            return True # overlap when curr is trend and prev is flat (with similar enough size), edge case scenario 11
        return False

    def has_partial_overlap_next(segment: dict, segment_next: dict) -> bool:
        """Checks whether overlap exists between curr & next, and current is more insignificant"""
        dir = segment['direction']
        start =  pd.to_datetime(segment['start'])
        end =  pd.to_datetime(segment['end'])
        width = (end - start).days

        next_dir = segment_next['direction']
        next_start = pd.to_datetime(segment_next['start'])
        next_end = pd.to_datetime(segment_next['end'])
        next_width = (next_end - next_start).days

        # Define conditions # TODO: Cleanup redunant condition statements no longer used.
        is_overlap_next = (end >= next_start)
        is_curr_shorter = (width <= next_width)
        is_next_noise = (next_dir == 'Noise') 
        is_trend_or_flat = (dir in ('Up', 'Down', 'Flat'))
        is_next_abrupt = ('trend_class' in segment_next and segment_next['trend_class'] == 'abrupt')

        if is_overlap_next and (is_trend_or_flat and (is_next_noise or is_next_abrupt) and not is_curr_shorter):
            return True # overlap when curr is trend and next is noise of larger window

        return False

    def has_partial_overlap_prev(segment: dict, segment_prev: dict) -> bool:
        """Light checks with overlaps on previous, that wouldnt already be covered by has_overlap_next"""
        dir = segment['direction']
        start =  pd.to_datetime(segment['start'])
        end =  pd.to_datetime(segment['end'])
        width = (end - start).days

        prev_dir = segment_prev['direction']
        prev_start = pd.to_datetime(segment_prev['start'])
        prev_end = pd.to_datetime(segment_prev['end'])
        prev_width = (prev_end - prev_start).days

        # Define conditions
        is_overlap_prev = (start <= prev_end)
        is_curr_shorter = (width <= prev_width)
        is_prev_noise = (prev_dir == 'Noise')
        is_trend_or_flat = (dir in ('Up', 'Down', 'Flat'))
        is_prev_abrupt = ('trend_class' in segment_prev and segment_prev['trend_class'] == 'abrupt')

        if is_overlap_prev and (is_trend_or_flat and (is_prev_noise or is_prev_abrupt) and not is_curr_shorter):
            return True # overlap when curr is trend and prev is noise of larger/equal window
        return False

    # Pass 1: Cleans inverse length segments in case any artifacts from expand/contract and abrupt shave logic
    segments = deepcopy(segments_refined)
    segments_refined = []
    for i, segment in enumerate(segments):
        if has_inverse(df, value_col, segment): 
            continue # Excludes segment.
        segments_refined.append(segment)

    if inverse_only:
        return segments_refined # stops early if only want Pass 1.

    # Pass 2: Cleans overlaps of same direction. Also artifacts from expansion/contraction & noise detection
    segments = deepcopy(segments_refined)
    segments_refined = [] 
    for i, segment in enumerate(segments):
        if (i < len(segments)-1 and has_overlap_next(segment, segments[i+1])) or \
            (i > 0 and has_overlap_prev(segment, segments[i-1])): 
            continue 
        segments_refined.append(segment)

    # Pass 3: Cleans partial overlaps with noise. Don't filter out completely when partial, adjust outside noise
    segments = deepcopy(segments_refined)
    segments_refined = [] 
    for i, segment in enumerate(segments):
        if (i < len(segments)-1 and has_partial_overlap_next(segment, segments[i+1])):

            shifted_end = (pd.to_datetime(segments[i+1]['start']) - pd.Timedelta(days=1))
            start = pd.to_datetime(segment['start'])
            is_inverted = (shifted_end < start) # In case noise segment is <= 1 day in length
            if is_inverted: 
                continue

            # when gradual, follows similar logic to expand/contract selection.
            end_df = df.loc[start:shifted_end]
            if segments[i]['direction'] == 'Up':
                new_end = end_df[value_col].idxmax()
                segments[i]['end'] = new_end.strftime('%Y-%m-%d')

            if segments[i]['direction'] == 'Down':
                new_end = end_df[value_col].idxmin()
                segments[i]['end'] = new_end.strftime('%Y-%m-%d')

            elif segments[i]['direction'] == 'Flat':
                segments[i]['end'] = shifted_end.strftime('%Y-%m-%d')

        if (i > 0 and has_partial_overlap_prev(segment, segments[i-1])): 

            shifted_start = (pd.to_datetime(segments[i-1]['end']) + pd.Timedelta(days=1))
            end = pd.to_datetime(segment['end'])

            # when gradual, follows similar logic to expand/contract selection.
            start_df = df.loc[shifted_start:end]
            if segments[i]['direction'] == 'Up':
                new_start = start_df[value_col].iloc[::-1].idxmin() + pd.Timedelta(days=1)
                segments[i]['start'] = new_start.strftime('%Y-%m-%d')

            if segments[i]['direction'] == 'Down':
                new_start = start_df[value_col].iloc[::-1].idxmax() + pd.Timedelta(days=1)
                segments[i]['start'] = new_start.strftime('%Y-%m-%d') 

            elif segments[i]['direction'] == 'Flat':
                segments[i]['start'] = shifted_start.strftime('%Y-%m-%d')

        segments_refined.append(segment)

    # Pass 4: Cleans inverse AGAIN: in case any artifacts from overlap adjustments
    segments = deepcopy(segments_refined)
    segments_refined = []
    for i, segment in enumerate(segments):
        if has_inverse(df, value_col, segment): 
            continue # Excludes segment.
        segments_refined.append(segment)

    # Pass 5: 
    # - Sets trends to noise when they have too low an SNR, too susceptible to noise, or not trendy enough
    # - Sets trends to flat when too flat.
    segments = deepcopy(segments_refined)
    segments_refined = [] 
    for i, segment in enumerate(segments):
        start = pd.to_datetime(segment['start'])
        end = pd.to_datetime(segment['end'])
        df_segment = df.loc[start:end].copy()

        # Conditions for edge cases
        left_is_noise = any(( # Consider segments within neighbour distance on left
                0 <= (start - pd.to_datetime(prev_seg['end'])).days <= GROUPING_DISTANCE
                and prev_seg.get('direction') == 'Noise'
            ) for k, prev_seg in enumerate(segments) if k != i)
        right_is_noise = any(( # Consider segments within neighbour distance on right
                0 <= (pd.to_datetime(next_seg['start']) - end).days <= GROUPING_DISTANCE
                and next_seg.get('direction') == 'Noise'
            ) for k, next_seg in enumerate(segments) if k != i)

        is_flat = segment['direction'] == 'Flat'
        is_gradual = ('trend_class' in segment and segment['trend_class'] == 'gradual')
        is_abrupt = ('trend_class' in segment and segment['trend_class'] == 'abrupt')
        is_padded = is_abrupt and ('padded' in segment) and (segment['padded'] == True)
        is_small = len(df_segment) <= 5

        # Edge case 1: Check SNR for trend but noise
        signal_power = np.mean(df_segment['signal']**2)
        noise_power = np.mean(df_segment['noise']**2)
        snr = float(10 * np.log10(signal_power / noise_power)) if noise_power != 0 else np.nan
        threshold_noise = 2.5 
        if is_gradual: threshold_noise = 5
        if is_flat: threshold_noise = 0
        too_noisy = (snr < threshold_noise)

        # Edge case 2.1: Check if abrupt segment near noise
        is_abrupt_near_noise = is_abrupt and (left_is_noise or right_is_noise)
        if is_padded: is_abrupt_near_noise = False # overwrite to False if segment got abrupt padded

        # Edge case 2.2: Check if gradual segment encapsulated by noise
        is_small_gradual_in_noise = is_gradual and (left_is_noise and right_is_noise) and is_small

        # Edge case 3.1: Check if value of end is too close to value of start
        value_start = df.loc[start, value_col]
        value_end = df.loc[end, value_col]
        diff = abs(value_end - value_start)
        threshold_diff = float(df['value_cleaned'].abs().max()) * 0.01
        if is_abrupt: # make a bit more lenient for abrupt
            threshold_diff = float(df['value_cleaned'].abs().max()) * 0.1
        trend_ends_too_close = (is_gradual or is_abrupt) and (diff <= threshold_diff)

        # Edge case 3.2: Check if total change too small, because noise puts it closer to 0
        total_change = abs(df_segment[value_col].diff().sum())
        threshold_diff = float(df['value_cleaned'].abs().max()) * 0.01
        trend_too_small = (is_gradual or is_abrupt) and (total_change <= threshold_diff)

        # Edge case 3.3: If max is not at end, or min is not at end for Up/Down trends - too flat for trend, consider as noise
        trend_too_flat = False
        if is_gradual and len(df_segment) >= 3:
            # Allow max/min to be in the last 30% of the segment instead of only at end
            segment_length = len(df_segment)
            last_30pct_start = int(segment_length * 0.7)
            last_section = df_segment.iloc[last_30pct_start:]

            if segment['direction'] == 'Up':
                max_date = df_segment[value_col].idxmax()
                max_in_last_section = (max_date in last_section.index)
                trend_too_flat = not max_in_last_section

            elif segment['direction'] == 'Down':
                min_date = df_segment[value_col].idxmin()
                min_in_last_section = (min_date in last_section.index)
                trend_too_flat = not min_in_last_section

        # Reclassify as noise if either edge cases met
        if too_noisy or (is_abrupt_near_noise and not trend_ends_too_close) or is_small_gradual_in_noise:
            segment['direction'] = 'Noise' 
            if 'trend_class' in segment: del segment['trend_class']

        if trend_ends_too_close or trend_too_small or trend_too_flat:
            segment['direction'] = 'Flat' 
            if 'trend_class' in segment: del segment['trend_class']

        segments_refined.append(segment)

    return segments_refined

fill_in_flats 

fill_in_flats(df, segments)

Fill uncovered time gaps with Flat segments using df's DateTimeIndex.

Adds Flat segments for: - Internal gaps between consecutive segments. - Leading gap before the first segment if df index starts earlier. - Trailing gap after the last segment if df index ends later.

Source code in pytrendy/post_processing/segments_refine/artifact_cleanup.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
def fill_in_flats(df: pd.DataFrame, segments: list[dict]) -> list[dict]:
    """Fill uncovered time gaps with Flat segments using df's DateTimeIndex.

    Adds Flat segments for:
    - Internal gaps between consecutive segments.
    - Leading gap before the first segment if df index starts earlier.
    - Trailing gap after the last segment if df index ends later.
    """
    if not segments: # if refinement produced no segments, cover full range as Flat and avoid index access errors below.
        start, end = df.index.min(), df.index.max()
        return [dict(start=start.strftime('%Y-%m-%d'), end=end.strftime('%Y-%m-%d'), direction='Flat')]

    segments_refined = segments.copy()

    # Leading gap
    data_start = df.index.min()
    first_start = pd.to_datetime(segments_refined[0]['start'])
    if data_start < first_start:
        lead_end = first_start - pd.Timedelta(days=1)
        if lead_end >= data_start:
            segments_refined.insert(0, dict(
                start=data_start.strftime('%Y-%m-%d'),
                end=lead_end.strftime('%Y-%m-%d'),
                direction='Flat'
            ))

    # Internal gaps (work on snapshot to avoid index shift confusion)
    j = 0
    original = segments_refined.copy()
    for i, curr_seg in enumerate(original):
        mapped = i + j
        if mapped >= len(segments_refined) - 1:
            continue
        next_seg = segments_refined[mapped + 1]

        gap_start = pd.to_datetime(curr_seg['end']) + pd.Timedelta(days=1)
        gap_end = pd.to_datetime(next_seg['start']) - pd.Timedelta(days=1)

        if gap_end >= gap_start:
            segments_refined.insert(mapped + 1, dict(
                start=gap_start.strftime('%Y-%m-%d'),
                end=gap_end.strftime('%Y-%m-%d'),
                direction='Flat'
            ))
            j += 1

    # Trailing gap
    data_end = df.index.max()
    last_end = pd.to_datetime(segments_refined[-1]['end'])
    if data_end > last_end:
        trail_start = last_end + pd.Timedelta(days=1)
        if data_end >= trail_start:
            segments_refined.append(dict(
                start=trail_start.strftime('%Y-%m-%d'),
                end=data_end.strftime('%Y-%m-%d'),
                direction='Flat'
            ))

    return segments_refined