Skip to content

results_pytrendy

Structured Access to Detection Results

PyTrendyResults 

PyTrendyResults(segments, index_type='date')

Wrapper class for accessing and analysing detected trend segments.

This class provides utilities for summarizing, filtering, and exporting trend segments detected by the detect_trends pipeline. It encapsulates both raw segment data and enhanced metrics such as rankings and signal-to-noise ratios.

Initializes the results object with a list of segments.

Parameters:

  • segments

    (list) –

    List of dictionaries representing individual trend segments.

Source code in pytrendy/io/results_pytrendy.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(self, segments: list[dict], index_type: str = 'date') -> None:
    """
    Initializes the results object with a list of segments.

    Args:
        segments (list):
            List of dictionaries representing individual trend segments.
        TODO Add desription for results type, if I want to go with this.
    """
    self.segments = segments
    self.trend_segments = [seg for seg in self.segments if 'trend_class' in seg] # Get segments that are trends (exclude flats and noise)

    self.index_type = index_type

    self.set_best()
    self.set_df()
    self.set_summary()

filter_segments 

filter_segments(direction='Any', sort_by='time_index', format='df')

Filters and sorts segments based on direction and ranking.

Parameters:

  • direction

    (str, default: 'Any' ) –

    Filter by trend direction. Options: 'Any', 'Up/Down', 'Up', 'Down', 'Flat', 'Noise'.

  • sort_by

    (str, default: 'time_index' ) –

    Sort segments by 'time_index' (ascending) or 'change_rank' (descending).

  • format

    (str, default: 'df' ) –

    Output format. 'df' returns a DataFrame, 'dict' returns a list of dictionaries.

Returns:

  • list | DataFrame –

    Union[list, pd.DataFrame]: Filtered and sorted segments in the specified format.

Source code in pytrendy/io/results_pytrendy.py 
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
def filter_segments(self, direction: str = 'Any', sort_by: str = 'time_index', format: str = 'df') -> list | pd.DataFrame:
    """
    Filters and sorts segments based on direction and ranking.

    Args:
        direction (str, optional):
            Filter by trend direction. Options: `'Any'`, `'Up/Down'`, `'Up'`, `'Down'`, `'Flat'`, `'Noise'`.
        sort_by (str, optional):
            Sort segments by `'time_index'` (ascending) or `'change_rank'` (descending).
        format (str, optional):
            Output format. `'df'` returns a DataFrame, `'dict'` returns a list of dictionaries.

    Returns:
        Union[`list`, `pd.DataFrame`]: Filtered and sorted segments in the specified format.

    """
    segments = self.segments
    if len(segments) == 0:
        return [] # return nothing if edge case

    # Sort segments by index/rank
    if sort_by == 'change_rank':
        segments = sorted(segments, key=lambda x: abs(x.get('total_change', 0)), reverse=True) # descending
    elif sort_by == 'time_index':
        segments = sorted(segments, key=lambda x: abs(x.get('time_index', 0))) # ascending
    else:
        print(f'{sort_by} is not a valid sort_by. Please try one of [\'time_index\', \'change_rank\']')

    # Filter segments by direction
    options = ['Any', 'Up/Down', 'Up', 'Down', 'Flat', 'Noise']
    if direction != 'Any' and direction in options:
        allowed_directions = {'Up', 'Down'} if direction == 'Up/Down' else {direction}
        segments = [seg for seg in segments if seg['direction'] in allowed_directions]
    if direction not in options:
        print(f'{direction} is not a valid direction. Please try one of [\'Any\', \'Up/Down\', \'Up\', \'Down\', \'Flat\', \'Noise\']')


    if format not in ['dict', 'df']:
        print(f'{format} is not a valid format. Please try one of [\'dict\', \'df\']')
    if format=='dict':
        return segments
    elif format == 'df':
        df = pd.DataFrame(segments)
        if len(segments) > 0:
            df = df.set_index('time_index')
        return df

    return segments

print_summary 

print_summary()

Prints a readable summary of detected trends.

Includes counts, best segment info, and a full tabular display.

Source code in pytrendy/io/results_pytrendy.py 
 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
def print_summary(self) -> None:
    """
    Prints a readable summary of detected trends.

    Includes counts, best segment info, and a full tabular display.
    """

    uptrends = self.summary['direction_counts']['Up'] if 'Up' in self.summary['direction_counts'] else 0
    downtrends = self.summary['direction_counts']['Down'] if 'Down' in self.summary['direction_counts'] else 0
    flats = self.summary['direction_counts']['Flat'] if 'Flat' in self.summary['direction_counts'] else 0 
    noise = self.summary['direction_counts']['Noise'] if 'Noise' in self.summary['direction_counts'] else 0 
    print(f'Detected: \n- {uptrends} Uptrends. \n- {downtrends} Downtrends.\n- {flats} Flats.\n- {noise} Noise.\n')

    descriptor = 'dates'
    if self.index_type in ['integer', 'float']:
        descriptor = 'indexes'
    elif self.index_type in ['string']:
        descriptor = 'labels'

    if len(self.filter_segments(direction='Up/Down')) == 0:
        print('Detected no trends...')
        return
    else:
        print(f'The best detected trend is {self.best["direction"]} between {descriptor} {self.best["start"]} - {self.best["end"]}\n')

    print('Full Results:')
    print('-------------------------------------------------------------------------------\n', 
          self.df_summary,
        '\n-------------------------------------------------------------------------------')

set_best 

set_best()

Identifies the best trend segment based on its total cumulative change, selecting the one with the lowest change rank.

1
2
3
- `results.best` returns best based on `total_change` (cumulative sum of differences). 
- Identifies the best trend segment based on steepness and duration.
- The segment with the lowest `change_rank` is selected as the best.
Source code in pytrendy/io/results_pytrendy.py 
35
36
37
38
39
40
41
42
43
44
45
46
def set_best(self) -> None:
    """
    Identifies the best trend segment based on its total cumulative change, selecting the one with the lowest change rank.

        - `results.best` returns best based on `total_change` (cumulative sum of differences). 
        - Identifies the best trend segment based on steepness and duration.
        - The segment with the lowest `change_rank` is selected as the best.
    """
    if len(self.trend_segments) == 0:
        self.best = None
        return
    self.best = min(self.trend_segments, key=lambda x: x.get('change_rank', math.inf))

set_df 

set_df()

Converts a list of trend segments into a pandas DataFrame for easier downstream analysis and data representation.

1
2
3
- Converts the segment list into a pandas `DataFrame`.
- Useful for downstream analysis and export.
- Alternative data representation to segments. In `dataframe` rather than `dict`
Source code in pytrendy/io/results_pytrendy.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def set_df(self) -> pd.DataFrame | None:
    """
    Converts a list of trend segments into a pandas DataFrame for easier downstream analysis and data representation.

        - Converts the segment list into a pandas `DataFrame`.
        - Useful for downstream analysis and export.
        - Alternative data representation to segments. In `dataframe` rather than `dict`
    """
    # Exit if nothing to report on
    if len(self.segments) == 0:
        return pd.DataFrame()

    df = pd.DataFrame(self.segments)
    df = df.set_index('time_index')
    self.df = df

set_summary 

set_summary()

Computes and stores summary statistics for trend segments, including a tabular overview and counts by direction.

1
2
- Computes summary statistics and stores a compact `DataFrame` of segments.
- Includes counts by direction and trend class, highest total change, and a tabular view.
Source code in pytrendy/io/results_pytrendy.py 
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
def set_summary(self) -> None:
    """
    Computes and stores summary statistics for trend segments, including a tabular overview and counts by direction.

        - Computes summary statistics and stores a compact `DataFrame` of segments.
        - Includes counts by direction and trend class, highest total change, and a tabular view.
    """
    summary = {}

    # Exit if nothing to report on
    if len(self.segments) == 0:
        summary['df'] = pd.DataFrame()
        return

    # Count the number of segments per direction type (Up, Down, Flat, Noise)
    direction_counts = Counter(seg["direction"] for seg in self.segments)
    summary["direction_counts"] = dict(direction_counts)

    # Count number of segments per trend classs (abrupt, gradual)
    trend_class_counts = Counter(seg["trend_class"] for seg in self.trend_segments)
    summary["trend_class_counts"] = dict(trend_class_counts)

    # Get array of total change from trends and get max (best) total change
    changes = [seg.get("total_change", 0) for seg in self.trend_segments]
    summary['highest_total_change'] = np.max(changes) if len(changes) > 0 else None

    # Set summary df (without extra details)
    df = pd.DataFrame(self.segments)

    unit_descriptor = {
        'date': 'days',
        'integer': 'index steps',
        'float': 'index steps',
        'string': 'index steps',
    }.get(self.index_type, 'days')

    df = df.rename({'days' : unit_descriptor}, axis = 1)

    cols = ['time_index', 'direction', 'start', 'end', unit_descriptor, 'total_change', 'change_rank']
    if len(changes) > 1:  #  only include trend_class if atleast one trend exists
        cols += ['trend_class']
    df = df[cols]

    df = df.set_index('time_index')
    self.df_summary = df

    # Set summary
    self.summary = summary