diff --git a/app/api/units_api.rb b/app/api/units_api.rb
index 8741ffb32..de8cb1d1b 100644
--- a/app/api/units_api.rb
+++ b/app/api/units_api.rb
@@ -578,6 +578,61 @@ class UnitsApi < Grape::API
     present unit.student_task_completion_stats, with: Grape::Presenters::Presenter
   end
 
+  desc 'Get historical task completion snapshots'
+  params do
+    optional :start_date, type: Date, desc: 'Include snapshots captured on or after this date'
+    optional :end_date, type: Date, desc: 'Include snapshots captured on or before this date'
+    optional :limit, type: Integer, desc: 'Maximum number of snapshots to return', default: 365
+  end
+  get '/units/:id/stats/task_completion_snapshots' do
+    unit = Unit.find(params[:id])
+    unless authorise? current_user, unit, :download_stats
+      error!({ error: "Not authorised to download stats of student tasks in #{unit.code}" }, 403)
+    end
+
+    snapshots = unit.task_completion_snapshots.order(snapshot_timestamp: :desc)
+    if params[:start_date].present?
+      start_timestamp = params[:start_date].in_time_zone.beginning_of_day.to_i
+      snapshots = snapshots.where('CAST(snapshot_timestamp AS UNSIGNED) >= ?', start_timestamp)
+    end
+    if params[:end_date].present?
+      end_timestamp = params[:end_date].in_time_zone.end_of_day.to_i
+      snapshots = snapshots.where('CAST(snapshot_timestamp AS UNSIGNED) <= ?', end_timestamp)
+    end
+    snapshots = snapshots.limit([params[:limit].to_i, 365].min)
+
+    present snapshots.map { |snapshot|
+      stats = snapshot.load_stats
+
+      {
+        snapshot_date: snapshot.snapshot_date,
+        snapshot_timestamp: snapshot.snapshot_timestamp,
+        stats: stats
+      }
+    }, with: Grape::Presenters::Presenter
+  end
+
+  desc 'Capture task completion snapshot immediately for this unit'
+  post '/units/:id/stats/task_completion_snapshots/capture' do
+    unit = Unit.find(params[:id])
+    unless authorise? current_user, unit, :capture_task_completion_snapshot
+      error!({ error: "Not authorised to capture stats of student tasks in #{unit.code}" }, 403)
+    end
+
+    # Check if a snapshot was captured within the past 30 minutes
+    recent_snapshot = unit.task_completion_snapshots.where('CAST(snapshot_timestamp AS UNSIGNED) > ?', 30.minutes.ago.to_i).order(snapshot_timestamp: :desc).first
+    if recent_snapshot.present?
+      recent_snapshot_time = recent_snapshot.snapshot_time
+      remaining_seconds = [(recent_snapshot_time + 30.minutes - Time.zone.now).ceil, 0].max
+      remaining_minutes = [(remaining_seconds / 60.0).ceil, 1].max
+      error!({ error: "A snapshot was captured at #{recent_snapshot_time.strftime('%H:%M')}. Please wait #{remaining_minutes} more minute(s) before capturing another snapshot." }, 429)
+    end
+
+    job_id = AggregateTaskCompletionStatsJob.perform_async(unit.id)
+    job = setup_job(job_id)
+    present job, with: Entities::SidekiqJobEntity
+  end
+
   desc 'Download stats related to the number of tasks assessed by each tutor'
   get '/csv/units/:id/tutor_assessments' do
     unit = Unit.find(params[:id])
diff --git a/app/helpers/file_helper.rb b/app/helpers/file_helper.rb
index 1f589f08a..a42c29bf4 100644
--- a/app/helpers/file_helper.rb
+++ b/app/helpers/file_helper.rb
@@ -276,6 +276,27 @@ def unit_portfolio_dir(unit, create: true, archived: true)
     dst
   end
 
+  def unit_analytics_dir(unit, create: true, archived: true)
+    dst = unit_work_root(unit, archived: archived)
+    dst << 'analytics/'
+
+    FileUtils.mkdir_p(dst) if create
+    dst
+  end
+
+  def unit_task_status_snapshots_dir(unit, create: true, archived: true)
+    dst = unit_analytics_dir(unit, create: create, archived: archived)
+    dst << 'task-statuses/'
+
+    FileUtils.mkdir_p(dst) if create
+    dst
+  end
+
+  def unit_task_status_snapshot_path(unit, snapshot_timestamp, create: true, archived: true)
+    snapshot_filename = "#{sanitized_filename(snapshot_timestamp.to_s)}.zip"
+    File.join(unit_task_status_snapshots_dir(unit, create: create, archived: archived), snapshot_filename)
+  end
+
   #
   # Generates a path for storing student portfolios
   #
@@ -778,6 +799,9 @@ def line_wrap(path, width: 160)
   module_function :unit_dir
   module_function :root_portfolio_dir
   module_function :unit_portfolio_dir
+  module_function :unit_analytics_dir
+  module_function :unit_task_status_snapshots_dir
+  module_function :unit_task_status_snapshot_path
   module_function :unit_work_root
   module_function :project_work_root
   module_function :student_portfolio_dir
diff --git a/app/models/task_completion_snapshot.rb b/app/models/task_completion_snapshot.rb
new file mode 100644
index 000000000..4b2468ac6
--- /dev/null
+++ b/app/models/task_completion_snapshot.rb
@@ -0,0 +1,110 @@
+# frozen_string_literal: true
+
+require 'csv'
+require 'zip'
+
+class TaskCompletionSnapshot < ApplicationRecord
+  include FileHelper
+
+  belongs_to :unit
+
+  validates :snapshot_timestamp, presence: true
+  validates :snapshot_timestamp, uniqueness: { scope: :unit_id }
+
+  after_destroy :delete_snapshot_file
+
+  def snapshot_file_path
+    FileHelper.unit_task_status_snapshot_path(unit, snapshot_timestamp, create: true)
+  end
+
+  def snapshot_contents
+    if File.exist?(snapshot_file_path)
+      return read_csv_from_zip(snapshot_file_path)
+    end
+    nil
+  rescue Zip::Error
+    nil
+  end
+
+  def snapshot_date
+    return nil if snapshot_timestamp.blank?
+
+    snapshot_time.to_date
+  end
+
+  def snapshot_time
+    return nil if snapshot_timestamp.blank?
+
+    Time.zone.at(snapshot_timestamp.to_i)
+  end
+
+  def load_stats
+    snapshot_contents = self.snapshot_contents
+
+    return {} if snapshot_contents.blank?
+
+    parse_csv_stats(snapshot_contents)
+  rescue CSV::MalformedCSVError
+    {}
+  end
+
+  def store_stats!(payload)
+    FileUtils.mkdir_p(File.dirname(snapshot_file_path))
+
+    tmp_path = "#{snapshot_file_path}.tmp"
+    Zip::OutputStream.open(tmp_path) do |zip|
+      zip.put_next_entry('snapshot.csv')
+      zip.write(payload.to_s)
+    end
+
+    FileUtils.mv(tmp_path, snapshot_file_path)
+  ensure
+    FileUtils.rm_f(tmp_path) if defined?(tmp_path)
+  end
+
+  private
+
+  def parse_csv_stats(csv_text)
+    csv = CSV.parse(csv_text, headers: true)
+    return {} if csv.empty?
+
+    stream_headers = unit.tutorial_streams.pluck(:abbreviation)
+    stream_headers = ['Tutorial'] if stream_headers.empty?
+    task_definitions = unit.task_definitions_by_grade
+
+    stats = Hash.new { |hash, key| hash[key] = Hash.new { |tutorial_hash, tutorial_key| tutorial_hash[tutorial_key] = Hash.new { |task_hash, task_key| task_hash[task_key] = Hash.new(0) } } }
+
+    csv.each do |row|
+      campus_abbreviation = row['Campus'].to_s.strip
+      next if campus_abbreviation.blank?
+
+      campus_name = Campus.find_by(abbreviation: campus_abbreviation)&.name || campus_abbreviation
+
+      stream_headers.each do |stream_header|
+        tutorial_name = row[stream_header].to_s.strip
+        next if tutorial_name.blank?
+
+        task_definitions.each do |task_definition|
+          status_value = row[task_definition.abbreviation].to_s.strip
+          status_key = TaskStatus.id_to_key(status_value.to_i) || :not_started
+          stats[campus_name][tutorial_name][task_definition.abbreviation][status_key.to_s] += 1
+        end
+      end
+    end
+
+    stats
+  end
+
+  def read_csv_from_zip(zip_path)
+    Zip::File.open(zip_path) do |zip_file|
+      entry = zip_file.find_entry('snapshot.csv') || zip_file.entries.first
+      return nil if entry.nil?
+
+      entry.get_input_stream.read
+    end
+  end
+
+  def delete_snapshot_file
+    FileUtils.rm_f(snapshot_file_path)
+  end
+end
diff --git a/app/models/unit.rb b/app/models/unit.rb
index c40d412e1..e5a7c4fe3 100644
--- a/app/models/unit.rb
+++ b/app/models/unit.rb
@@ -67,7 +67,8 @@ def self.permissions
       :get_tutor_times_summary,
       :get_marking_sessions,
       :upload_grades_csv,
-      :get_staff_notes
+      :get_staff_notes,
+      :capture_task_completion_snapshot
     ]
 
     # What can admin do with units?
@@ -157,6 +158,7 @@ def role_for(user)
   has_many :unit_roles, dependent: :destroy, inverse_of: :unit
   has_many :learning_outcomes, as: :context, dependent: :destroy # inverse_of: :unit
   has_many :marking_sessions, dependent: :destroy
+  has_many :task_completion_snapshots, dependent: :destroy, inverse_of: :unit
 
   has_many :comments, through: :projects
   has_many :tasks, through: :projects
@@ -1772,23 +1774,31 @@ def days_awaiting_feedback_by_tutorial_csv
   end
 
   def task_completion_csv
+    task_completion_csv_generator()
+  end
+
+  def task_completion_csv_generator(task_status_uses_id: false, includes_campus: false)
     task_def_by_grade = task_definitions_by_grade
     streams = tutorial_streams
     grp_sets = group_sets
+    base_headers = [
+      'Student ID',
+      'Username',
+      'Student Name',
+    ]
+    base_headers << 'Campus' if includes_campus
+    base_headers.push(
+      'Target Grade',
+      'Email',
+      'Portfolio',
+      'Grade',
+      'Rationale',
+      'Assessor',
+    )
 
     CSV.generate() do |csv|
       # Add header row
-      csv << ([
-        'Student ID',
-        'Username',
-        'Student Name',
-        'Target Grade',
-        'Email',
-        'Portfolio',
-        'Grade',
-        'Rationale',
-        'Assessor',
-      ] +
+      csv << (base_headers +
              (streams.count > 0 ? streams.map { |t| t.abbreviation } : ['Tutorial']) +
              grp_sets.map(&:name) +
              task_def_by_grade.map do |task_definition|
@@ -1803,7 +1813,11 @@ def task_completion_csv
       # Get the details to fetch for each task definition...
       td_select = task_def_by_grade.map do |td|
         result = []
-        result << "MAX(CASE WHEN tasks.task_definition_id = #{td.id} THEN (CASE WHEN task_statuses.name IS NULL THEN 'Not Started' ELSE task_statuses.name END) ELSE NULL END) AS status_#{td.id}"
+        if task_status_uses_id
+          result << "MAX(CASE WHEN tasks.task_definition_id = #{td.id} THEN (CASE WHEN tasks.task_status_id IS NULL THEN #{TaskStatus.not_started.id} ELSE tasks.task_status_id END) ELSE NULL END) AS status_#{td.id}"
+        else
+          result << "MAX(CASE WHEN tasks.task_definition_id = #{td.id} THEN (CASE WHEN task_statuses.name IS NULL THEN 'Not Started' ELSE task_statuses.name END) ELSE NULL END) AS status_#{td.id}"
+        end
         result << "MAX(CASE WHEN tasks.task_definition_id = #{td.id} THEN tasks.grade ELSE NULL END) AS grade_#{td.id}" if td.is_graded?
         result << "MAX(CASE WHEN tasks.task_definition_id = #{td.id} THEN tasks.quality_pts ELSE NULL END) AS stars_#{td.id}" if td.has_stars?
         result << "MAX(CASE WHEN tasks.task_definition_id = #{td.id} THEN tasks.contribution_pts ELSE NULL END) AS people_#{td.id}" if td.is_group_task?
@@ -1815,6 +1829,7 @@ def task_completion_csv
         .joins(
           :unit,
           'INNER JOIN users ON projects.user_id = users.id',
+          'LEFT OUTER JOIN campuses ON campuses.id = projects.campus_id',
           'INNER JOIN task_definitions ON task_definitions.unit_id = units.id',
           'LEFT OUTER JOIN tutorial_streams ON tutorial_streams.unit_id = units.id',
           'LEFT OUTER JOIN tutorial_enrolments ON tutorial_enrolments.project_id = projects.id',
@@ -1825,7 +1840,7 @@ def task_completion_csv
           'LEFT OUTER JOIN groups ON groups.id = group_memberships.group_id'
         ).select(
           'projects.id as project_id', 'users.student_id as student_id', 'users.username as username', 'users.first_name as first_name', 'projects.assessor_id as project_assessor',
-          'users.last_name as last_name', 'projects.target_grade', 'users.email as email', 'compile_portfolio', 'portfolio_production_date', 'grade', 'grade_rationale',
+          'users.last_name as last_name', 'campuses.abbreviation as campus_abbreviation', 'projects.target_grade', 'users.email as email', 'compile_portfolio', 'portfolio_production_date', 'grade', 'grade_rationale',
           *td_select,
           # Get tutorial for each stream in unit
           *streams.map { |s| "MAX(CASE WHEN tutorials.tutorial_stream_id = #{s.id} OR tutorials.tutorial_stream_id IS NULL THEN tutorials.abbreviation ELSE NULL END) AS tutorial_#{s.id}" },
@@ -1833,12 +1848,16 @@ def task_completion_csv
           "MAX(CASE WHEN tutorial_streams.id IS NULL THEN tutorials.abbreviation ELSE NULL END) AS tutorial",
           *grp_sets.map { |gs| "MAX(CASE WHEN groups.group_set_id = #{gs.id} THEN groups.name ELSE NULL END) AS grp_#{gs.id}" }
         ).group(
-          'projects.id', 'student_id', 'username', 'first_name', 'last_name', 'target_grade', 'email', 'compile_portfolio', 'portfolio_production_date', 'grade', 'grade_rationale'
+          'projects.id', 'student_id', 'username', 'first_name', 'last_name', 'campus_abbreviation', 'target_grade', 'email', 'compile_portfolio', 'portfolio_production_date', 'grade', 'grade_rationale'
         ).each do |row|
-          csv << ([
+          student_details = [
             row['student_id'],
             row['username'],
             "#{row['first_name']} #{row['last_name']}",
+          ]
+          student_details << row['campus_abbreviation'] if includes_campus
+
+          csv << (student_details + [
             GradeHelper.grade_for(row['target_grade']),
             row['email'],
             row['portfolio_production_date'].present? && !row['compile_portfolio'] && File.exist?(FileHelper.student_portfolio_path(self, row['username'], create: true)),
@@ -1854,7 +1873,11 @@ def task_completion_csv
           end.flatten + grp_sets.map do |gs|
             row["grp_#{gs.id}"]
           end + task_def_by_grade.map do |td|
-            result = [row["status_#{td.id}"].nil? ? TaskStatus.not_started.name : row["status_#{td.id}"]]
+            if task_status_uses_id
+              result = [row["status_#{td.id}"].nil? ? TaskStatus.not_started.id : row["status_#{td.id}"].to_i]
+            else
+              result = [row["status_#{td.id}"].nil? ? TaskStatus.not_started.name : row["status_#{td.id}"]]
+            end
             result << GradeHelper.short_grade_for(row["grade_#{td.id}"]) if td.is_graded?
             result << row["stars_#{td.id}"] if td.has_stars?
             result << row["people_#{td.id}"] if td.is_group_task?
@@ -3450,6 +3473,19 @@ def get_tutor_times_csv(start_date: nil, end_date: nil, timezone: nil, ignore_se
     end
   end
 
+  def capture_task_complete_stats_snapshot!(snapshot_time: Time.zone.now)
+    snapshot_payload = task_completion_csv_generator(task_status_uses_id: true, includes_campus: true)
+
+    timestamp = snapshot_time.to_i.to_s
+
+    task_completion_snapshots
+      .find_or_initialize_by(snapshot_timestamp: timestamp)
+      .tap do |snapshot|
+      snapshot.save!
+      snapshot.store_stats!(snapshot_payload)
+    end
+  end
+
   private
 
   def delete_associated_files
diff --git a/app/sidekiq/aggregate_task_completion_stats_job.rb b/app/sidekiq/aggregate_task_completion_stats_job.rb
new file mode 100644
index 000000000..4bb375ebe
--- /dev/null
+++ b/app/sidekiq/aggregate_task_completion_stats_job.rb
@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+class AggregateTaskCompletionStatsJob
+  include Sidekiq::Job
+  include Sidekiq::Status::Worker
+  include LogHelper
+  include ApplicationHelper
+
+  sidekiq_options lock: :until_executed,
+                  lock_args_method: ->(args) { [args.first] },
+                  on_conflict: :reject,
+                  retry: false
+
+  def perform(unit_id = nil)
+    logger.info 'Starting task completion stats aggregation...'
+
+    at(0)
+    total(1)
+
+    if unit_id.present?
+      Unit.find(unit_id).capture_task_complete_stats_snapshot!
+    else
+      Unit.active_units.find_each(&:capture_task_complete_stats_snapshot!)
+    end
+
+    at(1)
+    logger.info 'Completed task completion stats aggregation!'
+  rescue StandardError => e
+    logger.error e
+    raise e
+  end
+end
diff --git a/config/schedule.yml b/config/schedule.yml
index b0bd370f0..860c1d8a0 100644
--- a/config/schedule.yml
+++ b/config/schedule.yml
@@ -16,6 +16,10 @@ refresh_moderation_feedback_timestamps:
   cron: "every 60 minutes"
   class: "RefreshModerationFeedbackTimestampsJob"
 
+aggregate_task_completion_stats:
+  cron: "every day at 11:55pm"
+  class: "AggregateTaskCompletionStatsJob"
+
 # archive_old_units:
 #   cron: "every 6 months"
 #   class: "ArchiveOldUnitsJob"
diff --git a/db/migrate/20260322143502_create_task_completion_snapshots.rb b/db/migrate/20260322143502_create_task_completion_snapshots.rb
new file mode 100644
index 000000000..3cdf51c54
--- /dev/null
+++ b/db/migrate/20260322143502_create_task_completion_snapshots.rb
@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+class CreateTaskCompletionSnapshots < ActiveRecord::Migration[8.0]
+  def change
+    create_table :task_completion_snapshots do |t|
+      t.references :unit, null: false
+      t.string :snapshot_timestamp, null: false
+
+      t.timestamps
+    end
+
+    add_index :task_completion_snapshots, [:unit_id, :snapshot_timestamp], unique: true
+  end
+end
diff --git a/db/schema.rb b/db/schema.rb
index 1e49bb202..c6db56c98 100644
--- a/db/schema.rb
+++ b/db/schema.rb
@@ -405,6 +405,15 @@
     t.index ["user_id"], name: "index_task_comments_on_user_id"
   end
 
+  create_table "task_completion_snapshots", charset: "utf8mb4", collation: "utf8mb4_general_ci", force: :cascade do |t|
+    t.bigint "unit_id", null: false
+    t.string "snapshot_timestamp", null: false
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["unit_id", "snapshot_timestamp"], name: "idx_on_unit_id_snapshot_timestamp_e923c3ae10", unique: true
+    t.index ["unit_id"], name: "index_task_completion_snapshots_on_unit_id"
+  end
+
   create_table "task_definition_grade_due_dates", charset: "utf8mb4", collation: "utf8mb4_general_ci", force: :cascade do |t|
     t.bigint "task_definition_id", null: false
     t.integer "target_grade", null: false
diff --git a/test/api/units_api_test.rb b/test/api/units_api_test.rb
index 9ff3ed5d0..23add5b13 100644
--- a/test/api/units_api_test.rb
+++ b/test/api/units_api_test.rb
@@ -622,4 +622,169 @@ def test_draft_learning_summary_upload_requirements
     unit.reload
     assert_equal task_def_doc.id, unit.draft_task_definition_id
   end
+
+  def test_get_task_completion_snapshots
+    unit = FactoryBot.create :unit, with_students: false, task_count: 1, stream_count: 0, tutorials: 1, campus_count: 1
+    tutorial = unit.tutorials.first
+    task_definition = unit.task_definitions_by_grade.first
+
+    older_snapshot = TaskCompletionSnapshot.create!(
+      unit: unit,
+      snapshot_timestamp: Time.zone.parse('2026-04-01 10:00:00').to_i.to_s
+    )
+
+    mid_snapshot = TaskCompletionSnapshot.create!(
+      unit: unit,
+      snapshot_timestamp: Time.zone.parse('2026-04-02 10:00:00').to_i.to_s
+    )
+
+    latest_snapshot = TaskCompletionSnapshot.create!(
+      unit: unit,
+      snapshot_timestamp: Time.zone.parse('2026-04-03 10:00:00').to_i.to_s
+    )
+
+    older_snapshot.store_stats!(build_task_completion_snapshot_csv(tutorial, task_definition, [TaskStatus.not_started.id]))
+    mid_snapshot.store_stats!(build_task_completion_snapshot_csv(tutorial, task_definition, [TaskStatus.complete.id, TaskStatus.complete.id]))
+    latest_snapshot.store_stats!(build_task_completion_snapshot_csv(tutorial, task_definition, [TaskStatus.complete.id, TaskStatus.complete.id, TaskStatus.complete.id]))
+
+    add_auth_header_for(user: unit.main_convenor_user)
+    header 'Host', 'localhost'
+    get "/api/units/#{unit.id}/stats/task_completion_snapshots", { limit: 2 }
+
+    assert_equal 200, last_response.status, last_response_body
+    assert_equal 2, last_response_body.length
+
+    assert_equal latest_snapshot.snapshot_date.to_s, last_response_body[0]['snapshot_date'].to_date.to_s
+    assert_equal mid_snapshot.snapshot_date.to_s, last_response_body[1]['snapshot_date'].to_date.to_s
+
+    latest_stats = last_response_body[0]['stats']
+    assert_equal 3, latest_stats[tutorial.campus.name][tutorial.abbreviation][task_definition.abbreviation]['complete']
+
+    assert_not_equal older_snapshot.snapshot_date.to_s, last_response_body[1]['snapshot_date'].to_date.to_s
+  end
+
+  def test_get_task_completion_snapshots_filters_by_date
+    unit = FactoryBot.create :unit, with_students: false, task_count: 1, stream_count: 0, tutorials: 1, campus_count: 1
+    tutorial = unit.tutorials.first
+    task_definition = unit.task_definitions_by_grade.first
+
+    TaskCompletionSnapshot.create!(
+      unit: unit,
+      snapshot_timestamp: Time.zone.parse('2026-03-30 10:00:00').to_i.to_s
+    )
+
+    included_snapshot = TaskCompletionSnapshot.create!(
+      unit: unit,
+      snapshot_timestamp: Time.zone.parse('2026-04-02 10:00:00').to_i.to_s
+    )
+
+    TaskCompletionSnapshot.create!(
+      unit: unit,
+      snapshot_timestamp: Time.zone.parse('2026-04-05 10:00:00').to_i.to_s
+    )
+
+    unit.task_completion_snapshots.find_each do |snapshot|
+      snapshot.store_stats!(build_task_completion_snapshot_csv(tutorial, task_definition, [TaskStatus.complete.id]))
+    end
+
+    add_auth_header_for(user: unit.main_convenor_user)
+    header 'Host', 'localhost'
+    get "/api/units/#{unit.id}/stats/task_completion_snapshots", {
+      start_date: Date.new(2026, 4, 1),
+      end_date: Date.new(2026, 4, 3)
+    }
+
+    assert_equal 200, last_response.status, last_response_body
+    assert_equal 1, last_response_body.length
+    assert_equal included_snapshot.snapshot_date.to_s, last_response_body[0]['snapshot_date'].to_date.to_s
+  end
+
+  def test_get_task_completion_snapshots_not_authorised
+    unit = FactoryBot.create :unit, with_students: false, task_count: 0
+    TaskCompletionSnapshot.create!(
+      unit: unit,
+      snapshot_timestamp: Time.zone.now.to_i.to_s
+    )
+
+    add_auth_header_for(user: User.where(role: Role.student).first)
+    header 'Host', 'localhost'
+    get "/api/units/#{unit.id}/stats/task_completion_snapshots"
+
+    assert_equal 403, last_response.status
+  end
+
+  def test_post_capture_task_completion_snapshot
+    Sidekiq::Testing.inline! do
+      unit = FactoryBot.create :unit
+
+      count_before = TaskCompletionSnapshot.where(unit: unit).count
+
+      add_auth_header_for(user: unit.main_convenor_user)
+      header 'Host', 'localhost'
+      post "/api/units/#{unit.id}/stats/task_completion_snapshots/capture"
+
+      assert_equal 201, last_response.status, last_response_body
+      assert_not_nil last_response_body['id']
+
+      snapshot = TaskCompletionSnapshot.where(unit: unit).order(snapshot_timestamp: :desc).first
+      assert_not_nil snapshot
+      assert_equal count_before + 1, TaskCompletionSnapshot.where(unit: unit).count
+
+      assert_equal Date.current.to_s, snapshot.snapshot_date.to_s
+      assert_not_empty snapshot.load_stats
+      assert File.exist?(snapshot.snapshot_file_path)
+    ensure
+      Sidekiq::Testing.fake!
+    end
+  end
+
+  def test_post_capture_task_completion_snapshot_not_authorised
+    unit = FactoryBot.create :unit, with_students: false, task_count: 0
+
+    add_auth_header_for(user: User.where(role: Role.student).first)
+    header 'Host', 'localhost'
+    post "/api/units/#{unit.id}/stats/task_completion_snapshots/capture"
+
+    assert_equal 403, last_response.status
+  end
+
+  private
+
+  def build_task_completion_snapshot_csv(tutorial, task_definition, statuses)
+    headers = [
+      'Student ID',
+      'Username',
+      'Student Name',
+      'Campus',
+      'Target Grade',
+      'Email',
+      'Portfolio',
+      'Grade',
+      'Rationale',
+      'Assessor',
+      'Tutorial',
+      task_definition.abbreviation,
+    ]
+
+    CSV.generate do |csv|
+      csv << headers
+
+      statuses.each_with_index do |status, index|
+        csv << [
+          "#{index + 1}",
+          "student-#{index + 1}",
+          "Student #{index + 1}",
+          tutorial.campus.abbreviation,
+          '0',
+          "student-#{index + 1}@example.com",
+          'false',
+          '',
+          '',
+          '',
+          tutorial.abbreviation,
+          status,
+        ]
+      end
+    end
+  end
 end
diff --git a/test/factories/task_completion_snapshot_factory.rb b/test/factories/task_completion_snapshot_factory.rb
new file mode 100644
index 000000000..871bd0611
--- /dev/null
+++ b/test/factories/task_completion_snapshot_factory.rb
@@ -0,0 +1,6 @@
+FactoryBot.define do
+  factory :task_completion_snapshot do
+    unit
+    snapshot_timestamp { Time.current.to_i.to_s }
+  end
+end
diff --git a/test/models/task_completion_snapshot_test.rb b/test/models/task_completion_snapshot_test.rb
new file mode 100644
index 000000000..38300242b
--- /dev/null
+++ b/test/models/task_completion_snapshot_test.rb
@@ -0,0 +1,109 @@
+require 'test_helper'
+
+class TaskCompletionSnapshotTest < ActiveSupport::TestCase
+  setup do
+    @unit = FactoryBot.create(:unit, with_students: false, task_count: 1, stream_count: 0, tutorials: 1, campus_count: 1)
+    @snapshot = FactoryBot.create(:task_completion_snapshot, unit: @unit)
+  end
+
+  test 'task_completion_snapshot belongs to unit' do
+    assert @snapshot.unit.is_a?(Unit)
+    assert_equal @unit, @snapshot.unit
+  end
+
+  test 'task_completion_snapshot is valid with all attributes' do
+    assert @snapshot.valid?
+  end
+
+  test 'task_completion_snapshot is invalid without snapshot_timestamp' do
+    snapshot = FactoryBot.build(:task_completion_snapshot, snapshot_timestamp: nil)
+    assert_not snapshot.valid?
+    assert snapshot.errors[:snapshot_timestamp].include?("can't be blank")
+  end
+
+  test 'task_completion_snapshot enforces unique snapshot_timestamp per unit' do
+    duplicate = FactoryBot.build(
+      :task_completion_snapshot,
+      unit: @unit,
+      snapshot_timestamp: @snapshot.snapshot_timestamp
+    )
+
+    assert_not duplicate.valid?
+    assert duplicate.errors[:snapshot_timestamp].include?('has already been taken')
+  end
+
+  test 'task_completion_snapshot allows same snapshot_timestamp for different units' do
+    other_unit = FactoryBot.create(:unit)
+    snapshot = FactoryBot.build(
+      :task_completion_snapshot,
+      unit: other_unit,
+      snapshot_timestamp: @snapshot.snapshot_timestamp
+    )
+
+    assert snapshot.valid?
+  end
+
+  test 'store_stats! writes a csv file contained in a zip that can be loaded' do
+    tutorial = @unit.tutorials.first
+    task_definition = @unit.task_definitions_by_grade.first
+
+    payload = CSV.generate do |csv|
+      csv << ['Student ID', 'Username', 'Student Name', 'Campus', 'Target Grade', 'Email', 'Portfolio', 'Grade', 'Rationale', 'Assessor', 'Tutorial', task_definition.abbreviation]
+      csv << ['1', 'student-1', 'Student 1', tutorial.campus.abbreviation, '0', 'student-1@example.com', 'false', '', '', '', tutorial.abbreviation, TaskStatus.complete.id]
+      csv << ['2', 'student-2', 'Student 2', tutorial.campus.abbreviation, '0', 'student-2@example.com', 'false', '', '', '', tutorial.abbreviation, TaskStatus.complete.id]
+      csv << ['3', 'student-3', 'Student 3', tutorial.campus.abbreviation, '0', 'student-3@example.com', 'false', '', '', '', tutorial.abbreviation, TaskStatus.complete.id]
+      csv << ['4', 'student-4', 'Student 4', tutorial.campus.abbreviation, '0', 'student-4@example.com', 'false', '', '', '', tutorial.abbreviation, TaskStatus.complete.id]
+    end
+
+    expected = {
+      tutorial.campus.name => {
+        tutorial.abbreviation => {
+          task_definition.abbreviation => {
+            'complete' => 4
+          }
+        }
+      }
+    }
+
+    @snapshot.store_stats!(payload)
+
+    assert File.exist?(@snapshot.snapshot_file_path)
+    assert_equal expected, @snapshot.load_stats
+  end
+
+  test 'load_stats returns empty hash if file missing' do
+    snapshot = FactoryBot.create(
+      :task_completion_snapshot,
+      unit: @unit,
+      snapshot_timestamp: (Time.zone.now.to_i + 100).to_s
+    )
+
+    FileUtils.rm_f(snapshot.snapshot_file_path)
+    assert_equal({}, snapshot.load_stats)
+  end
+
+  test 'deleting snapshot deletes associated zip file' do
+    tutorial = @unit.tutorials.first
+    task_definition = @unit.task_definitions_by_grade.first
+
+    payload = CSV.generate do |csv|
+      csv << ['Student ID', 'Username', 'Student Name', 'Target Grade', 'Email', 'Portfolio', 'Grade', 'Rationale', 'Assessor', 'Tutorial', task_definition.abbreviation]
+      csv << ['1', 'student-1', 'Student 1', '0', 'student-1@example.com', 'false', '', '', '', tutorial.abbreviation, TaskStatus.complete.id]
+    end
+
+    @snapshot.store_stats!(payload)
+
+    file_path = @snapshot.snapshot_file_path
+    assert File.exist?(file_path)
+
+    @snapshot.destroy
+    assert_not File.exist?(file_path)
+  end
+
+  test 'snapshot_date is derived from snapshot_timestamp' do
+    timestamp = Time.zone.local(2026, 4, 8, 23, 55, 0).to_i.to_s
+    snapshot = FactoryBot.build(:task_completion_snapshot, snapshot_timestamp: timestamp)
+
+    assert_equal Date.new(2026, 4, 8), snapshot.snapshot_date
+  end
+end
diff --git a/test/models/unit_model_test.rb b/test/models/unit_model_test.rb
index 6a54c9b28..23e03b411 100644
--- a/test/models/unit_model_test.rb
+++ b/test/models/unit_model_test.rb
@@ -626,6 +626,34 @@ def test_task_completion_csv_all_td_in_one_stream
     check_task_completion_csv unit
   end
 
+  def test_task_completion_csv_generator_includes_campus_when_requested
+    unit = FactoryBot.create :unit, campus_count: 2, tutorials: 2, stream_count: 1, task_count: 1, student_count: 5, unenrolled_student_count: 0, part_enrolled_student_count: 0
+
+    csv_str = unit.task_completion_csv_generator(includes_campus: true)
+    CSV.parse(csv_str,
+              headers: true,
+              return_headers: false,
+              header_converters: [->(body) { body&.encode('UTF-8', 'binary', invalid: :replace, undef: :replace, replace: '')&.downcase }],
+              converters: [->(body) { body&.encode('UTF-8', 'binary', invalid: :replace, undef: :replace, replace: '') }]).each do |entry|
+      user = User.find_by(username: entry['username'])
+      assert user.present?, entry.inspect
+
+      project = unit.active_projects.find_by(user_id: user.id)
+      assert project.present?, entry.inspect
+
+      assert_equal project.campus&.abbreviation, entry['campus'], entry.inspect
+    end
+  end
+
+  def test_task_completion_csv_generator_excludes_campus_by_default
+    unit = FactoryBot.create :unit, campus_count: 2, tutorials: 1, stream_count: 1, task_count: 1, student_count: 2, unenrolled_student_count: 0, part_enrolled_student_count: 0
+
+    csv_str = unit.task_completion_csv_generator
+    parsed = CSV.parse(csv_str, headers: true)
+
+    assert_not_includes parsed.headers.map(&:downcase), 'campus'
+  end
+
   def test_export_users
     unit = FactoryBot.create :unit, campus_count: 2, tutorials:2, stream_count:0, task_count:3, student_count:8, unenrolled_student_count: 0, part_enrolled_student_count: 0, set_one_of_each_task: true
 
@@ -1138,4 +1166,122 @@ def test_cant_disable_aip_only_while_aip_tasks_exist
     assert_includes unit.errors[:mark_late_submissions_as_assess_in_portfolio], 'cannot be disabled while tasks are in the Assess in Portfolio state'
   end
 
+
+  test 'capture-task-complete-stats-snapshot creates snapshot for date' do
+    data = build_unit_with_controlled_task_statuses
+    unit = data[:unit]
+    snapshot_time = Time.zone.local(2026, 4, 8, 23, 55, 0)
+    expected_stats = parse_task_completion_stats_csv(unit, unit.task_completion_csv_generator(task_status_uses_id: true, includes_campus: true))
+
+    count_before = unit.task_completion_snapshots.count
+    snapshot = unit.capture_task_complete_stats_snapshot!(snapshot_time: snapshot_time)
+
+    assert_equal count_before + 1, unit.task_completion_snapshots.count
+    assert_equal snapshot_time.to_date, snapshot.snapshot_date
+    assert_equal snapshot_time.to_i.to_s, snapshot.snapshot_timestamp
+    assert_equal expected_stats, snapshot.load_stats
+
+    persisted_snapshot = unit.task_completion_snapshots.find_by(snapshot_timestamp: snapshot_time.to_i.to_s)
+    assert_not_nil persisted_snapshot
+    assert_equal snapshot.id, persisted_snapshot.id
+  ensure
+    unit&.destroy
+  end
+
+  test 'capture-task-complete-stats-snapshot creates a new snapshot for a new timestamp' do
+    data = build_unit_with_controlled_task_statuses
+    unit = data[:unit]
+    task_definitions = data[:task_definitions]
+    student2 = data[:student2]
+
+    first_time = Time.zone.local(2026, 4, 8, 9, 0, 0)
+    second_time = Time.zone.local(2026, 4, 8, 20, 0, 0)
+
+    first_snapshot = unit.capture_task_complete_stats_snapshot!(snapshot_time: first_time)
+    first_stats = first_snapshot.load_stats.deep_dup
+    count_before = unit.task_completion_snapshots.count
+
+    # Change one task status so the new capture has different stats.
+    student2.task_for_task_definition(task_definitions[0]).update!(task_status: TaskStatus.fail)
+    expected_updated_stats = parse_task_completion_stats_csv(unit, unit.task_completion_csv_generator(task_status_uses_id: true, includes_campus: true))
+
+    updated_snapshot = unit.capture_task_complete_stats_snapshot!(snapshot_time: second_time)
+
+    assert_equal count_before + 1, unit.task_completion_snapshots.count
+    assert_not_equal first_snapshot.id, updated_snapshot.id
+    assert_equal second_time.to_i.to_s, updated_snapshot.snapshot_timestamp
+    assert_not_equal first_stats, updated_snapshot.load_stats
+    assert_equal expected_updated_stats, updated_snapshot.load_stats
+  ensure
+    unit&.destroy
+  end
+
+  private
+
+  def parse_task_completion_stats_csv(unit, csv_text)
+    csv = CSV.parse(csv_text, headers: true)
+    streams = unit.tutorial_streams.pluck(:abbreviation)
+    streams = ['Tutorial'] if streams.empty?
+    task_definitions = unit.task_definitions_by_grade
+    campus_header = csv.headers.find { |header| header.to_s.casecmp('Campus').zero? }
+
+    campus_names_by_abbreviation = if campus_header.present?
+                                     abbreviations = csv.map { |row| row[campus_header].to_s.strip }.reject(&:blank?).uniq
+                                     Campus.where(abbreviation: abbreviations).pluck(:abbreviation, :name).to_h
+                                   else
+                                     {}
+                                   end
+
+    csv.each_with_object(Hash.new { |hash, key| hash[key] = {} }) do |row, stats|
+      streams.each do |stream_name|
+        tutorial_name = row[stream_name].to_s.strip
+        next if tutorial_name.blank?
+
+        campus_abbreviation = campus_header.present? ? row[campus_header].to_s.strip : nil
+
+        campus_name = if campus_abbreviation.present?
+                        campus_names_by_abbreviation[campus_abbreviation] || campus_abbreviation
+                      elsif stream_name == 'Tutorial'
+                        unit.tutorials.find_by(abbreviation: tutorial_name)&.campus&.name || stream_name
+                      else
+                        stream_name
+                      end
+
+        stats[campus_name][tutorial_name] ||= {}
+
+        task_definitions.each do |task_definition|
+          status_name = row[task_definition.abbreviation].to_s.strip
+          status_key = TaskStatus.id_to_key(status_name.to_i) || :not_started
+          stats[campus_name][tutorial_name][task_definition.abbreviation] ||= Hash.new(0)
+          stats[campus_name][tutorial_name][task_definition.abbreviation][status_key.to_s] += 1
+        end
+      end
+    end
+  end
+
+  def build_unit_with_controlled_task_statuses
+    unit = FactoryBot.create(:unit, with_students: false, task_count: 2, stream_count: 0, tutorials: 1, campus_count: 1)
+    tutorial = unit.tutorials.first
+    campus = tutorial.campus
+    task_definitions = unit.task_definitions.order(:id).to_a
+
+    student1 = unit.enrol_student(FactoryBot.create(:user, :student), campus)
+    student2 = unit.enrol_student(FactoryBot.create(:user, :student), campus)
+    student1.enrol_in(tutorial)
+    student2.enrol_in(tutorial)
+
+    student1.task_for_task_definition(task_definitions[0]).update!(task_status: TaskStatus.complete)
+    student2.task_for_task_definition(task_definitions[0]).update!(task_status: TaskStatus.complete)
+    student1.task_for_task_definition(task_definitions[1]).update!(task_status: TaskStatus.fail)
+    student2.task_for_task_definition(task_definitions[1]).update!(task_status: TaskStatus.not_started)
+
+    {
+      unit: unit,
+      tutorial: tutorial,
+      task_definitions: task_definitions,
+      student1: student1,
+      student2: student2
+    }
+  end
+
 end
diff --git a/test/sidekiq/scheduled_job_test.rb b/test/sidekiq/scheduled_job_test.rb
index ba1b1b7e0..2100284ba 100644
--- a/test/sidekiq/scheduled_job_test.rb
+++ b/test/sidekiq/scheduled_job_test.rb
@@ -6,13 +6,14 @@ class TiiCheckProgressJobTest < ActiveSupport::TestCase
   def test_jobs_are_scheduled
     Sidekiq::Cron::Job.destroy_all!
     Sidekiq::Cron::Job.load_from_hash!(YAML.load_file(Rails.root.join('config/schedule.yml')))
-    assert_equal 4, Sidekiq::Cron::Job.all.count, Sidekiq::Cron::Job.all.map(&:name)
+    assert_equal 5, Sidekiq::Cron::Job.all.count, Sidekiq::Cron::Job.all.map(&:name)
 
     Sidekiq::Cron::Job.all.each(&:enqueue!)
     assert_equal 1, TiiRegisterWebHookJob.jobs.count
     assert_equal 1, TiiCheckProgressJob.jobs.count
     assert_equal 1, ClearAccessTokensJob.jobs.count
     assert_equal 1, RefreshModerationFeedbackTimestampsJob.jobs.count
+    assert_equal 1, AggregateTaskCompletionStatsJob.jobs.count
     # assert_equal 1, ArchiveOldUnitsJob.jobs.count
   end