LinkedIn-Analytics-Dashboard/growth.py

"""Tracks follower-growth history over time: an append-only-per-day log that powers the
sidebar Growth Statistics indicator, the Growth tab's historical table/chart, and the PDF
report's weekly-pace section. Persisted as JSON in UPLOAD_DIR so it survives restarts and
uploads on any team member's machine, not just the original author's.

Pace is judged against a rolling personal best rather than a fixed percentage: the current
week's growth-to-date is compared against the best of the last ROLLING_BEST_WEEKS COMPLETED
weeks (never against earlier days within the current week itself). Weeks are real Monday-
Sunday calendar weeks, not a daily trailing window and not a 7-day block anchored to
whichever day-of-week the data happens to start on — an earlier version compared each day's
trailing-7-day rate against the best trailing rate from the past N weeks, which could
include days from later in the SAME displayed week, so a week could get flagged "Behind" due
to its own mid-week peak rather than genuinely falling short of prior weeks. This rises
automatically as performance improves, but a one-off spike ages out of the window after
ROLLING_BEST_WEEKS instead of setting an unreachable target forever.

Zero import-time dependency on app.py, for the same reason report.py has none: app.py runs
as "__main__" (`python3 app.py`), so `import app` here would re-execute it as a second
module copy that's still mid-import when these functions get called.
"""
import datetime
import json
import os

import pandas as pd

ROLLING_BEST_WEEKS = 4


def _history_path(upload_dir):
    return os.path.join(upload_dir, "growth_history.json")


def load_history(upload_dir):
    path = _history_path(upload_dir)
    if not os.path.exists(path):
        return []
    with open(path) as f:
        return json.load(f)


def _save_history(upload_dir, history):
    with open(_history_path(upload_dir), "w") as f:
        json.dump(history, f, indent=2)


def _followers_as_of(fts_df, on_or_before):
    """Cumulative Followers value from the most recent row on or before `on_or_before`;
    falls back to the earliest available row if the whole series starts later."""
    prior = fts_df[fts_df["Date"] <= on_or_before]
    if prior.empty:
        return float(fts_df["Cumulative Followers"].iloc[0])
    return float(prior["Cumulative Followers"].iloc[-1])


def _week_start_for(date):
    """Monday of the standard calendar week `date` falls into — real Monday-Sunday weeks,
    not a 7-day block anchored to whatever day-of-week the data happens to start on (which
    produced Tuesday-Monday "weeks" here, since the tracked history starts on a Tuesday)."""
    return date - pd.Timedelta(days=date.weekday())


def compute_snapshot(fts_df, prior_history, week_boundary_date=None):
    """Computes (but does not persist) a growth snapshot as of the last date in fts_df,
    given the history entries recorded so far (used for since_last_pct and the rolling-best
    pace target). Taking the prior history as a plain list — rather than reading it from
    disk — lets backfill_history() replay this against progressively larger slices of
    fts_df without hitting disk each time.

    `week_boundary_date` only affects which past weeks are eligible for the rolling-best
    lookback below — it does NOT move week-to-date's own week, which always stays the
    actual last date in fts_df (so a week with zero new data yet never gets falsely reset
    to a hypothetical 0%). Defaults to this entry's own week, which reproduces the original
    behavior of excluding the current week from its own target pool. backfill_history()
    passes today's real date for the latest entry only, so a week that's actually finished
    in real life (e.g. LinkedIn's export lagging a day behind on Mondays) immediately
    becomes eligible as a rolling-best candidate, instead of waiting for next week's data
    to start arriving before it's ever counted."""
    fts_df = fts_df.sort_values("Date")
    today = fts_df["Date"].iloc[-1]
    total_followers = float(fts_df["Cumulative Followers"].iloc[-1])

    # Trailing 7-day rate: a smoother, purely informational "how's it going right now"
    # figure. Not used for the pace judgment (see week_to_date_pct below) since a trailing
    # window updates every day and doesn't respect week boundaries.
    week_ago = today - pd.Timedelta(days=7)
    followers_week_ago = _followers_as_of(fts_df, week_ago)
    weekly_pct = ((total_followers - followers_week_ago) / followers_week_ago * 100
                  if followers_week_ago else None)
    weekly_pct = round(weekly_pct, 2) if weekly_pct is not None else None

    if prior_history:
        prev_total = prior_history[-1]["total_followers"]
        since_last_pct = (total_followers - prev_total) / prev_total * 100 if prev_total else None
    else:
        since_last_pct = None

    # Discrete week-over-week pace: growth since the start of THIS week (always the real
    # last-data week — never moved by week_boundary_date), compared against the best of the
    # last ROLLING_BEST_WEEKS COMPLETED weeks.
    current_week_start = _week_start_for(today)
    week_start_baseline = _followers_as_of(fts_df, current_week_start - pd.Timedelta(days=1))
    week_to_date_pct = ((total_followers - week_start_baseline) / week_start_baseline * 100
                         if week_start_baseline else 0.0)
    week_to_date_pct = round(week_to_date_pct, 2)

    # Which past weeks are eligible for the rolling-best pool: strictly before the lookback
    # boundary, which defaults to this entry's own week (excluding it from its own target,
    # since an ONGOING week can't fairly be judged against its own still-accumulating
    # total) but can be pushed forward to the real current date for the live/latest entry —
    # at that point this week is no longer "ongoing" in real life even if fresh data for it
    # hasn't arrived, so it's fair game as a candidate for whatever comes next.
    lookback_boundary_week_start = (_week_start_for(week_boundary_date)
                                     if week_boundary_date is not None else current_week_start)
    lookback_start = lookback_boundary_week_start - pd.Timedelta(weeks=ROLLING_BEST_WEEKS)
    last_entry_per_week = {}
    for h in prior_history:
        h_week_start = h.get("week_start")
        if h_week_start is None or not (lookback_start <= pd.Timestamp(h_week_start) < lookback_boundary_week_start):
            continue
        existing = last_entry_per_week.get(h_week_start)
        if existing is None or h["date"] > existing["date"]:
            last_entry_per_week[h_week_start] = h
    completed_week_rates = [h["week_to_date_pct"] for h in last_entry_per_week.values()]
    rolling_best_weekly_pct = round(max(completed_week_rates), 2) if completed_week_rates else None
    on_track = (None if rolling_best_weekly_pct is None
                else week_to_date_pct >= rolling_best_weekly_pct)

    return {
        "date": today.strftime("%Y-%m-%d"),
        "total_followers": int(total_followers),
        "weekly_pct": weekly_pct,
        "since_last_pct": round(since_last_pct, 2) if since_last_pct is not None else None,
        "week_start": current_week_start.strftime("%Y-%m-%d"),
        "week_to_date_pct": week_to_date_pct,
        "rolling_best_weekly_pct": rolling_best_weekly_pct,
        "rolling_best_weeks": ROLLING_BEST_WEEKS,
        "on_track": on_track,
    }


def last_two_completed_weeks(history):
    """The most recent COMPLETED week's final entry, and the one before it — both full
    Monday-Sunday weeks. Used for an apples-to-apples week-over-week comparison, as opposed
    to comparing a full week against the current week, which is always partial until the
    next Monday's upload.

    "Completed" is judged against the real calendar (today's actual date), not the last date
    present in the export — LinkedIn's export routinely lags a day or two behind (e.g. it may
    stop at Saturday), which previously made an already-finished week look "still in
    progress" and made the sidebar fall back to showing the week before it."""
    if not history:
        return None, None
    real_today = pd.Timestamp(datetime.date.today())
    current_week_start = (real_today - pd.Timedelta(days=real_today.weekday())).strftime("%Y-%m-%d")
    last_entry_per_week = {}
    for h in history:
        if h["week_start"] >= current_week_start:
            continue
        existing = last_entry_per_week.get(h["week_start"])
        if existing is None or h["date"] > existing["date"]:
            last_entry_per_week[h["week_start"]] = h
    completed_weeks = sorted(last_entry_per_week.values(), key=lambda h: h["week_start"], reverse=True)
    last_week = completed_weeks[0] if len(completed_weeks) >= 1 else None
    prior_week = completed_weeks[1] if len(completed_weeks) >= 2 else None
    return last_week, prior_week


def backfill_history(fts_df, upload_dir):
    """Ensures the growth-history log has one entry per day already present in fts_df —
    backfills any days from before this feature existed (each computed using only the data
    that would have been available up to that day, so backfilled entries are historically
    accurate, not hindsight-adjusted), while always recomputing the MOST RECENT day fresh so
    a same-day re-upload/correction updates that day's entry instead of reusing a stale one.
    Returns the full merged history list, sorted chronologically."""
    existing = {e["date"]: e for e in load_history(upload_dir)}
    fts_sorted = fts_df.sort_values("Date")
    dates = list(fts_sorted["Date"])
    real_today = pd.Timestamp(datetime.date.today())
    history = []
    for i, d in enumerate(dates):
        date_str = d.strftime("%Y-%m-%d")
        is_latest = (i == len(dates) - 1)
        if date_str in existing and not is_latest:
            history.append(existing[date_str])
            continue
        subset = fts_sorted[fts_sorted["Date"] <= d]
        # Only the latest entry gets the real-calendar boundary — backfilled days stay
        # reproducible, using just their own date as before.
        boundary = max(d, real_today) if is_latest else None
        history.append(compute_snapshot(subset, history, week_boundary_date=boundary))
    _save_history(upload_dir, history)
    return history