Get direction weight changes compared to previous period Args: user_id: User ID days: Days to look back Returns: List of direction changes with delta information
(user_id: str, days: int = 7)
| 1222 | |
| 1223 | |
| 1224 | def get_direction_changes(user_id: str, days: int = 7) -> List[Dict]: |
| 1225 | """ |
| 1226 | Get direction weight changes compared to previous period |
| 1227 | |
| 1228 | Args: |
| 1229 | user_id: User ID |
| 1230 | days: Days to look back |
| 1231 | |
| 1232 | Returns: |
| 1233 | List of direction changes with delta information |
| 1234 | """ |
| 1235 | current_profile = get_profile(user_id) |
| 1236 | previous_profile = get_profile_snapshot(user_id, days) |
| 1237 | |
| 1238 | if not current_profile or not previous_profile: |
| 1239 | return [] |
| 1240 | |
| 1241 | current_directions = current_profile.get("core_directions", {}) |
| 1242 | previous_directions = previous_profile.get("core_directions", {}) |
| 1243 | |
| 1244 | changes = [] |
| 1245 | all_directions = set(current_directions.keys()) | set(previous_directions.keys()) |
| 1246 | |
| 1247 | for direction in all_directions: |
| 1248 | current_weight = current_directions.get(direction, 0.0) |
| 1249 | previous_weight = previous_directions.get(direction, 0.0) |
| 1250 | delta = current_weight - previous_weight |
| 1251 | |
| 1252 | changes.append({ |
| 1253 | "direction": direction, |
| 1254 | "current_weight": current_weight, |
| 1255 | "previous_weight": previous_weight, |
| 1256 | "delta": delta, |
| 1257 | "trend": "up" if delta > 0.001 else ("down" if delta < -0.001 else "stable") |
| 1258 | }) |
| 1259 | |
| 1260 | changes.sort(key=lambda x: abs(x["delta"]), reverse=True) |
| 1261 | return changes |
| 1262 | |
| 1263 | |
| 1264 | # ============== Recent Push Operations ============== |
no test coverage detected