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 | def plot_pytrendy(df: pd.DataFrame, value_col: str, segments_enhanced: list[dict], index_type: str = "date", suppress_show: bool = False) -> plt.Figure:
"""
Visualizes detected trend segments over the original time series signal.
This function overlays shaded regions on the signal to indicate trends such as Up, Down, Flat, and Noise
It also annotates ranked segments and handles visual adjustments for abrupt transitions.
Args:
df (pd.DataFrame):
Time series data with datetime index and signal column.
value_col (str):
Name of the column containing the signal to plot.
segments_enhanced (list):
List of segment dictionaries containing keys like `'start'`, `'end'`, `'direction'`, `'trend_class'`, and `'change_rank'`.
index_type (str):
The type of index passed by the user. Different index types require different logic. Currently Accepted Index Types are: "date", "integer", "float".
suppress_show (bool, optional):
If True, suppresses the automatic display of the plot with plt.show(). Defaults to False.
Returns:
matplotlib.figure.Figure:
The figure object containing the plot. Can be displayed with `plt.show()` or saved.
"""
# Define colours
color_map = {
'Up': 'lightgreen',
'Down': 'lightcoral',
'Flat': 'lightblue',
'Noise': 'lightgray',
}
accepted_index_types = ['date', 'datetime64', 'integer', 'float', 'string']
if index_type not in accepted_index_types:
raise NotImplementedError(f"Index Type {index_type} not yet implemented.")
fig, ax = plt.subplots(figsize=(20, 5))
# Plot the value line
ax.plot(df.index, df[value_col], color='black', lw=1)
# Add shaded regions with fill_between
ymin, ymax = ax.get_ylim() # get plot's visible y-range
for i, seg in enumerate(segments_enhanced):
if index_type == "date":
start = pd.to_datetime(seg['start'])
end = pd.to_datetime(seg['end'])
else:
start = seg['start']
end = seg['end']
color = color_map.get(seg['direction'], 'gray')
# Get context on prev seg if possible
prev_seg = segments_enhanced[i-1] if i-1 >= 0 else None
if index_type == "date":
prev_neighbouring = prev_seg and (pd.to_datetime(prev_seg['end']) == (start - pd.Timedelta(days=1)))
elif index_type == 'string':
prev_neighbouring = prev_seg and (prev_seg['end'] == df.index[df.index.get_loc(start) - 1])
else:
prev_neighbouring = prev_seg and (prev_seg['end'] == (start - 1))
is_prev_not_trend = prev_seg and (not ('trend_class' in prev_seg))
# Current seg context
is_abrupt = ('trend_class' in seg and seg['trend_class'] == 'abrupt')
is_noise = (seg['direction'] == 'Noise')
is_not_trend = not ('trend_class' in seg)
# Get context on next seg if possible
next_seg = segments_enhanced[i+1] if i+1 < len(segments_enhanced) else None
if index_type == 'date':
next_neighbouring = next_seg and (pd.to_datetime(next_seg['start']) == (end + pd.Timedelta(days=1)))
elif index_type == 'string':
end_pos = df.index.get_loc(end)
next_neighbouring = next_seg and (next_seg['start'] == _safe_adjacent(df.index, end_pos, 1))
else:
next_neighbouring = next_seg and (next_seg['start'] == (end + 1))
next_seg_abrupt = next_seg and (('trend_class' in next_seg) and (next_seg['trend_class'] == 'abrupt'))
next_seg_noise = next_seg and (next_seg['direction'] == 'Noise')
# Adjust starts when appropriate
if is_abrupt or is_noise:
pass # Keep start as-is for abrupt/noise segments
else:
if index_type == 'date':
new_start = start - pd.Timedelta(days=1) # Everything else displaced left start
elif index_type == 'string':
start_pos = df.index.get_loc(start)
new_start = _safe_adjacent(df.index, start_pos, -1)
else:
new_start = start - 1 # Everything else displaced left start
# Check validity of plot start adjustment
value_new_start = df.loc[new_start, value_col] if new_start is not None and new_start in df.index else None
value = df.loc[start, value_col]
valid_up_start = (value_new_start) and (seg['direction'] == 'Up') and (value_new_start < value)
valid_down_start = (value_new_start) and (seg['direction'] == 'Down') and (value_new_start > value)
if valid_up_start or valid_down_start or is_not_trend:
start = new_start # Apply left displacement only if valid
else:
# if not displaced and prev is not trend, adjust by plotting (as prev has already been drawn)
if is_prev_not_trend and prev_neighbouring:
if index_type == 'date':
prev_end = pd.to_datetime(segments_enhanced[i-1]['end'])
prev_new_end = (prev_end + pd.Timedelta(days=1)).strftime('%Y-%m-%d')
elif index_type == 'string':
prev_end = segments_enhanced[i-1]['end']
prev_end_pos = df.index.get_loc(prev_end)
prev_new_end = _safe_adjacent(df.index, prev_end_pos, 1)
else:
prev_end = segments_enhanced[i-1]['end']
prev_new_end = prev_end + 1
if prev_new_end is not None:
if index_type == 'string':
mask = (np.arange(len(df)) >= df.index.get_loc(prev_end)) & (np.arange(len(df)) <= df.index.get_loc(prev_new_end))
else:
mask = (df.index >= prev_end) & (df.index <= prev_new_end)
prev_color = color_map.get(segments_enhanced[i-1]['direction'], 'gray')
ax.fill_between(df.index[mask], ymin, ymax, color=prev_color, alpha=0.4)
# Adjust ends when appropriate
if (next_seg_abrupt or next_seg_noise) and next_neighbouring:
if index_type == 'date':
new_end = end + pd.Timedelta(days=1)
elif index_type == 'string':
end_pos = df.index.get_loc(end)
new_end = _safe_adjacent(df.index, end_pos, 1)
else:
new_end = end + 1
# Check validity of plot end adjustment
value_new_end = df.loc[new_end, value_col] if new_end is not None and new_end in df.index else None
value = df.loc[end, value_col]
valid_up_end = (value_new_end) and (seg['direction'] == 'Up') and (value_new_end > value)
valid_down_end = (value_new_end) and (seg['direction'] == 'Down') and (value_new_end < value)
is_not_trend = not ('trend_class' in seg)
if valid_up_end or valid_down_end or is_not_trend:
end = new_end # Apply right displacement only if valid
else:
# if not displaced and next is noise, adjust for next plotting round
if next_seg_noise and next_neighbouring:
if index_type == 'date':
segments_enhanced[i+1]['start'] = (pd.to_datetime(segments_enhanced[i+1]['start']) - pd.Timedelta(days=1)).strftime('%Y-%m-%d')
elif index_type == 'string':
next_start_pos = df.index.get_loc(segments_enhanced[i+1]['start'])
segments_enhanced[i+1]['start'] = _safe_adjacent(df.index, next_start_pos, -1)
else:
segments_enhanced[i+1]['start'] = (segments_enhanced[i+1]['start'] - 1)
else:
pass # Keep end as-is
if index_type == 'string':
mask = (np.arange(len(df)) >= df.index.get_loc(start)) & (np.arange(len(df)) <= df.index.get_loc(end))
else:
mask = (df.index >= start) & (df.index <= end)
ax.fill_between(df.index[mask], ymin, ymax, color=color, alpha=0.4)
# Add ranking if up/down trend
if 'change_rank' in seg and seg['direction'] in ['Up', 'Down']:
if index_type in ['string']:
midpoint = int((df.index.get_loc(end) - df.index.get_loc(start))/2)
mid_date = df.index[df.index.get_loc(start) + midpoint]
else:
mid_date = start + (end - start) / 2
y_pos = ymax - (ymax - ymin) * 0.05
ax.text(mid_date, y_pos, str(seg['change_rank']), fontsize=12,
fontweight='bold', ha='center', va='top',
color=color[5:])
# Add vertical line if next seg is same & touching
if next_seg and next_neighbouring and next_seg['direction'] == seg['direction']:
if index_type == 'date':
line_date = pd.to_datetime(seg['end'])
else:
line_date = seg['end']
ax.axvline(x=line_date, color=color[5:], linewidth=0.5)
# Set limits
if index_type == 'string':
first_date = df.index[0]
last_date = df.index[-1]
else:
first_date = df.index.min()
last_date = df.index.max()
ax.set_xlim(first_date, last_date)
ax.set_ylim(ymin, ymax)
if index_type == 'date':
# Major ticks: every 7 days (with labels)
ax.xaxis.set_major_locator(mdates.WeekdayLocator(interval=1))
ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d'))
# Minor ticks: every day (no labels, just tick marks/grid)
ax.xaxis.set_minor_locator(mdates.DayLocator())
# Rotate major tick labels
plt.setp(ax.get_xticklabels(), rotation=90, ha='right')
# Optional: show grid lines for both
ax.grid(True, which='major', color='gray', alpha=0.3)
if index_type == 'string':
ticks = ax.get_xticks()
labels = [t.get_text() for t in ax.get_xticklabels()]
n = 10
ax.set_xticks(ticks[::n])
ax.set_xticklabels(labels[::n], rotation=90, ha='center')
ax.set_title("PyTrendy Detection", fontsize=20)
if index_type == 'date':
ax.set_xlabel("Date")
elif index_type == 'string':
ax.set_xlabel('Label')
else:
ax.set_xlabel("Index")
ax.set_ylabel("Value")
# Create custom legend handles (colored boxes)
legend_handles = [
mpatches.Patch(color='lightgreen', alpha=0.4, label='Up'),
mpatches.Patch(color='lightcoral', alpha=0.4, label='Down'),
mpatches.Patch(color='lightblue', alpha=0.4, label='Flat'),
mpatches.Patch(color='lightgray', alpha=0.4, label='Noise'),
]
ax.legend(handles=legend_handles, loc='upper right',
bbox_to_anchor=(1, 1.15), ncol=4, frameon=True)
plt.tight_layout()
if not suppress_show:
plt.show()
return fig
|