diff --git a/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/purchase/__init__.py b/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/purchase/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/purchase/api.py b/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/purchase/api.py
index 5092f54e4..1bd82c528 100644
--- a/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/purchase/api.py
+++ b/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/purchase/api.py
@@ -1,11 +1,12 @@
 # ruff: noqa: N801, N815 invalid-name
+from marshmallow import ValidationError, validates_schema
+from marshmallow.fields import Dict, List, String
+
 from cc_common.data_model.schema.base_record import ForgivingSchema
 from cc_common.data_model.schema.compact.api import CompactOptionsResponseSchema
 from cc_common.data_model.schema.compact.common import COMPACT_TYPE
 from cc_common.data_model.schema.jurisdiction.api import JurisdictionOptionsResponseSchema
 from cc_common.data_model.schema.jurisdiction.common import JURISDICTION_TYPE
-from marshmallow import ValidationError, validates_schema
-from marshmallow.fields import Dict, List, String
 
 
 class PurchasePrivilegeOptionsResponseSchema(ForgivingSchema):
diff --git a/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/transaction/__init__.py b/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/transaction/__init__.py
index e69de29bb..389e55633 100644
--- a/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/transaction/__init__.py
+++ b/backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/transaction/__init__.py
@@ -0,0 +1,72 @@
+# ruff: noqa: N802 we use camelCase to match the marshmallow schema definition
+
+from cc_common.data_model.schema.common import CCDataClass
+from cc_common.data_model.schema.transaction.record import TransactionRecordSchema
+
+
+class TransactionData(CCDataClass):
+    """
+    Class representing a Transaction with read-only properties.
+
+    Unlike several other CCDataClass subclasses, this one does not include setters. This is because
+    transaction records are only created during transaction processing, so we can pass the entire record
+    from the processing into the constructor.
+
+    Note: This class requires valid data when created - it cannot be instantiated empty
+    and populated later.
+    """
+
+    # Define the record schema at the class level
+    _record_schema = TransactionRecordSchema()
+
+    # Require valid data when creating instances
+    _requires_data_at_construction = True
+
+    @property
+    def transactionProcessor(self) -> str:
+        return self._data['transactionProcessor']
+
+    @property
+    def transactionId(self) -> str:
+        return self._data['transactionId']
+
+    @property
+    def batch(self) -> dict:
+        """Batch information containing batchId, settlementState, settlementTimeLocal, and settlementTimeUTC."""
+        return self._data['batch']
+
+    @property
+    def lineItems(self) -> list[dict]:
+        """
+        List of line items, each containing description, itemId, name, quantity, taxable,
+        unitPrice, and optionally privilegeId.
+        """
+        return self._data['lineItems']
+
+    @property
+    def compact(self) -> str:
+        return self._data['compact']
+
+    @property
+    def licenseeId(self) -> str:
+        return self._data['licenseeId']
+
+    @property
+    def responseCode(self) -> str:
+        return self._data['responseCode']
+
+    @property
+    def settleAmount(self) -> str:
+        return self._data['settleAmount']
+
+    @property
+    def submitTimeUTC(self) -> str:
+        return self._data['submitTimeUTC']
+
+    @property
+    def transactionStatus(self) -> str:
+        return self._data['transactionStatus']
+
+    @property
+    def transactionType(self) -> str:
+        return self._data['transactionType']
diff --git a/backend/compact-connect/lambdas/python/common/cc_common/data_model/transaction_client.py b/backend/compact-connect/lambdas/python/common/cc_common/data_model/transaction_client.py
index 069852438..275a230b2 100644
--- a/backend/compact-connect/lambdas/python/common/cc_common/data_model/transaction_client.py
+++ b/backend/compact-connect/lambdas/python/common/cc_common/data_model/transaction_client.py
@@ -3,7 +3,8 @@
 from boto3.dynamodb.conditions import Key
 
 from cc_common.config import _Config, logger
-from cc_common.data_model.schema.transaction.record import TransactionRecordSchema, UnsettledTransactionRecordSchema
+from cc_common.data_model.schema.transaction import TransactionData
+from cc_common.data_model.schema.transaction.record import UnsettledTransactionRecordSchema
 
 AUTHORIZE_DOT_NET_CLIENT_TYPE = 'authorize.net'
 
@@ -14,21 +15,18 @@ class TransactionClient:
     def __init__(self, config: _Config):
         self.config = config
 
-    def store_transactions(self, transactions: list[dict]) -> None:
+    def store_transactions(self, transactions: list[TransactionData]) -> None:
         """
         Store transaction records in DynamoDB.
 
-        :param compact: The compact name
         :param transactions: List of transaction records to store
         """
         with self.config.transaction_history_table.batch_writer() as batch:
             for transaction in transactions:
                 # Convert UTC timestamp to epoch for sorting
-                transaction_processor = transaction['transactionProcessor']
+                transaction_processor = transaction.transactionProcessor
                 if transaction_processor == AUTHORIZE_DOT_NET_CLIENT_TYPE:
-                    transaction_schema = TransactionRecordSchema()
-
-                    serialized_record = transaction_schema.dump(transaction)
+                    serialized_record = transaction.serialize_to_database_record()
                     batch.put_item(Item=serialized_record)
                 else:
                     raise ValueError(f'Unsupported transaction processor: {transaction_processor}')
@@ -80,6 +78,49 @@ def get_transactions_in_range(self, compact: str, start_epoch: int, end_epoch: i
 
         return all_items
 
+    def get_most_recent_transaction_for_compact(self, compact: str) -> TransactionData:
+        """
+        Get the most recent transaction for a compact.
+
+        Starts by querying the current month's partition key (based on config.current_standard_datetime),
+        then sequentially queries previous months until a record is found.
+
+        :param compact: The compact name
+        :return: The most recent transaction for the compact
+        :raises ValueError: If no transactions are found for the compact
+        """
+        # Start with the current month
+        current_date = self.config.current_standard_datetime.replace(day=1)
+        # During normal operations, the most recent transaction should be no more than two days old, if there were any
+        # transactions in that period. We'll look back up to three months, which should cover most reasonable
+        # situations.
+        max_months_to_check = 3
+
+        for _ in range(max_months_to_check):
+            month_key = current_date.strftime('%Y-%m')
+            pk = f'COMPACT#{compact}#TRANSACTIONS#MONTH#{month_key}'
+
+            # Query for the most recent transaction in this month (descending order, limit 1)
+            response = self.config.transaction_history_table.query(
+                KeyConditionExpression=Key('pk').eq(pk),
+                ScanIndexForward=False,  # Descending order (most recent first)
+                Limit=1,
+            )
+
+            items = response.get('Items', [])
+            if items:
+                # Found a transaction, return it
+                return TransactionData.from_database_record(items[0])
+
+            # Move to previous month
+            if current_date.month == 1:
+                current_date = current_date.replace(year=current_date.year - 1, month=12)
+            else:
+                current_date = current_date.replace(month=current_date.month - 1)
+
+        # No transactions found after checking max_months_to_check months
+        raise ValueError(f'No transactions found for compact: {compact}')
+
     def _query_transactions_for_month(
         self,
         compact: str,
@@ -122,10 +163,13 @@ def _query_transactions_for_month(
 
     def _set_privilege_id_in_line_item(self, line_items: list[dict], item_id_prefix: str, privilege_id: str):
         for line_item in line_items:
-            if line_item.get('itemId').lower().startswith(item_id_prefix.lower()):
+            item_id = line_item.get('itemId')
+            if item_id and item_id.lower().startswith(item_id_prefix.lower()):
                 line_item['privilegeId'] = privilege_id
 
-    def add_privilege_information_to_transactions(self, compact: str, transactions: list[dict]) -> list[dict]:
+    def add_privilege_information_to_transactions(
+        self, compact: str, transactions: list[TransactionData]
+    ) -> list[TransactionData]:
         """
         Add privilege and licensee IDs to transaction line items based on the jurisdiction they were purchased for.
 
@@ -134,7 +178,7 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
         :return: Modified list of transactions with privilege and licensee IDs added to line items
         """
         for transaction in transactions:
-            line_items = transaction['lineItems']
+            line_items = transaction.lineItems
             # Extract jurisdictions from line items with format priv:{compact}-{jurisdiction}-{license type abbr}
             jurisdictions_to_process = set()
             for line_item in line_items:
@@ -145,7 +189,7 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
                     jurisdictions_to_process.add(jurisdiction)
 
             # Query for privilege records using the GSI
-            gsi_pk = f'COMPACT#{compact}#TX#{transaction["transactionId"]}#'
+            gsi_pk = f'COMPACT#{compact}#TX#{transaction.transactionId}#'
             response = self.config.provider_table.query(
                 IndexName=self.config.compact_transaction_id_gsi_name,
                 KeyConditionExpression=Key('compactTransactionIdGSIPK').eq(gsi_pk),
@@ -157,9 +201,9 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
                 logger.error(
                     'No privilege records found for this transaction id.',
                     compact=compact,
-                    transaction_id=transaction['transactionId'],
+                    transaction_id=transaction.transactionId,
                     # attempt to grab the licensee id from the authorize.net data, which may be invalid if it was masked
-                    licensee_id=transaction['licenseeId'],
+                    licensee_id=transaction.licenseeId,
                 )
                 # We mark the data as UNKNOWN so it still shows up in the history,
                 # and move onto the next transaction
@@ -182,16 +226,16 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
                 logger.error(
                     'More than one matching provider id found for a transaction id.',
                     compact=compact,
-                    transaction_id=transaction['transactionId'],
+                    transaction_id=transaction.transactionId,
                     # attempt to grab the licensee id from the authorize.net data, which may be invalid if it was masked
-                    provider_ids=transaction['licenseeId'],
+                    provider_ids=transaction.licenseeId,
                 )
 
             # The licensee id recorded in Authorize.net cannot be trusted, as Authorize.net masks any values that look
             # like a credit card number (consecutive digits separated by dashes). We need to grab the provider id from
             # the privileges associated with this transaction and set the licensee id on the transaction to that value
             # to ensure it is valid.
-            transaction['licenseeId'] = provider_ids.pop()
+            transaction.update({'licenseeId': provider_ids.pop()})
 
             # Process each privilege record
             for jurisdiction in jurisdictions_to_process:
@@ -220,15 +264,16 @@ def add_privilege_information_to_transactions(self, compact: str, transactions:
                         'No matching jurisdiction privilege record found for transaction. '
                         'Cannot determine privilege id for this transaction',
                         compact=compact,
-                        transactionId=transaction['transactionId'],
+                        transactionId=transaction.transactionId,
                         jurisdiction=jurisdiction,
-                        provider_id=transaction['licenseeId'],
+                        provider_id=transaction.licenseeId,
                         matching_privilege_records=response.get('Items', []),
                     )
                     # we set the privilege id to UNKNOWN, so that it will be visible in the report
                     self._set_privilege_id_in_line_item(
                         line_items=line_items, item_id_prefix=item_id_prefix, privilege_id='UNKNOWN'
                     )
+            transaction.update({'lineItems': line_items})
 
         return transactions
 
@@ -269,7 +314,7 @@ def store_unsettled_transaction(self, compact: str, transaction_id: str, transac
                 error=str(e),
             )
 
-    def reconcile_unsettled_transactions(self, compact: str, settled_transactions: list[dict]) -> list[str]:
+    def reconcile_unsettled_transactions(self, compact: str, settled_transactions: list[TransactionData]) -> list[str]:
         """
         Reconcile unsettled transactions with settled transactions and detect old unsettled transactions.
 
@@ -297,7 +342,7 @@ def reconcile_unsettled_transactions(self, compact: str, settled_transactions: l
             return []
 
         # Create a set of settled transaction IDs for efficient lookup
-        settled_transaction_ids = {tx['transactionId'] for tx in settled_transactions}
+        settled_transaction_ids = {tx.transactionId for tx in settled_transactions}
 
         # Separate matched and unmatched unsettled transactions
         matched_unsettled = []
@@ -325,6 +370,16 @@ def reconcile_unsettled_transactions(self, compact: str, settled_transactions: l
         cutoff_time = datetime.now(UTC) - timedelta(hours=48)
         old_unsettled_transactions = []
 
+        # We expect that all transactions we process from Authorize.net will match a record we have already
+        # created at the time of purchase, as an unsettled transaction. Any mismatch is an error.
+        matched_unsettled_transaction_ids = {tx['transactionId'] for tx in matched_unsettled}
+        unmatched_settled_transaction_ids = settled_transaction_ids - matched_unsettled_transaction_ids
+        if unmatched_settled_transaction_ids:
+            logger.error(
+                'Unable to reconcile some transactions from Authorize.Net with our unsettled transactions',
+                unreconciled_transactions=unmatched_settled_transaction_ids
+            )
+
         for unsettled_tx in unmatched_unsettled:
             transaction_date = datetime.fromisoformat(unsettled_tx['transactionDate'])
             if transaction_date < cutoff_time:
diff --git a/backend/compact-connect/lambdas/python/common/common_test/test_constants.py b/backend/compact-connect/lambdas/python/common/common_test/test_constants.py
index 54043c5c3..d88206c9b 100644
--- a/backend/compact-connect/lambdas/python/common/common_test/test_constants.py
+++ b/backend/compact-connect/lambdas/python/common/common_test/test_constants.py
@@ -29,6 +29,37 @@
 DEFAULT_PRIVILEGE_EXPIRATION_DATE = '2025-04-04'
 DEFAULT_PRIVILEGE_UPDATE_DATETIME = '2020-05-05T12:59:59+00:00'
 DEFAULT_COMPACT_TRANSACTION_ID = '1234567890'
+DEFAULT_COMPACT_TRANSACTION_BATCH = {
+    'batchId': '67890',
+    'settlementState': 'settledSuccessfully',
+    'settlementTimeLocal': '2024-01-01T09:00:00',
+    'settlementTimeUTC': '2024-01-01T13:00:00.000Z',
+}
+DEFAULT_COMPACT_TRANSACTION_PRIVILEGE_LINE_ITEM = {
+    'description': 'Compact Privilege for Ohio',
+    'itemId': 'priv:aslp-oh',
+    'name': 'Ohio Compact Privilege',
+    'quantity': '1.0',
+    'taxable': 'False',
+    'unitPrice': '100.00',
+    'privilegeId': 'mock-privilege-id-oh',
+}
+DEFAULT_COMPACT_TRANSACTION_COMPACT_LINE_ITEM = {
+    'description': 'Compact fee applied for each privilege purchased',
+    'itemId': 'aslp-compact-fee',
+    'name': 'ASLP Compact Fee',
+    'quantity': '1',
+    'taxable': 'False',
+    'unitPrice': '10.50',
+}
+DEFAULT_COMPACT_TRANSACTION_FEE_LINE_ITEM = {
+    'description': 'credit card transaction fee',
+    'itemId': 'credit-card-transaction-fee',
+    'name': 'Credit Card Transaction Fee',
+    'quantity': '1',
+    'taxable': 'False',
+    'unitPrice': '3.00',
+}
 DEFAULT_PRIVILEGE_ID = 'SLP-NE-1'
 DEFAULT_MILITARY_AFFILIATION_TYPE = 'militaryMember'
 DEFAULT_MILITARY_STATUS = 'active'
@@ -57,6 +88,7 @@
 PRIVILEGE_RECORD_TYPE = 'privilege'
 PRIVILEGE_UPDATE_RECORD_TYPE = 'privilegeUpdate'
 PROVIDER_RECORD_TYPE = 'provider'
+TRANSACTION_RECORD_TYPE = 'transaction'
 
 # Privilege update default values
 DEFAULT_PRIVILEGE_UPDATE_TYPE = 'renewal'
diff --git a/backend/compact-connect/lambdas/python/common/common_test/test_data_generator.py b/backend/compact-connect/lambdas/python/common/common_test/test_data_generator.py
index fd2e36e83..883056a9f 100644
--- a/backend/compact-connect/lambdas/python/common/common_test/test_data_generator.py
+++ b/backend/compact-connect/lambdas/python/common/common_test/test_data_generator.py
@@ -1,5 +1,6 @@
 # ruff: noqa: F403, F405 star import of test constants file
 import json
+from copy import deepcopy
 from datetime import date, datetime
 from decimal import Decimal
 
@@ -648,6 +649,38 @@ def put_default_jurisdiction_configuration_in_configuration_table(
 
         return jurisdiction_config
 
+    @staticmethod
+    def generate_default_transaction(value_overrides: dict | None = None):
+        """Generate a default transaction"""
+        from cc_common.data_model.schema.transaction import TransactionData
+
+        # We'll fill in any missing batch values with defaults
+        default_batch = deepcopy(DEFAULT_COMPACT_TRANSACTION_BATCH)
+        if value_overrides and 'batch' in value_overrides.keys():
+            default_batch.update(value_overrides.pop('batch'))
+
+        default_transaction = {
+            'transactionProcessor': 'authorize.net',
+            'transactionId': DEFAULT_COMPACT_TRANSACTION_ID,
+            'batch': default_batch,
+            'lineItems': [
+                DEFAULT_COMPACT_TRANSACTION_PRIVILEGE_LINE_ITEM,
+                DEFAULT_COMPACT_TRANSACTION_COMPACT_LINE_ITEM,
+                DEFAULT_COMPACT_TRANSACTION_FEE_LINE_ITEM,
+            ],
+            'compact': DEFAULT_COMPACT,
+            'licenseeId': DEFAULT_PROVIDER_ID,
+            'responseCode': '1',
+            'settleAmount': '113.50',
+            'submitTimeUTC': '2024-01-01T12:00:00.000Z',
+            'transactionStatus': 'settledSuccessfully',
+            'transactionType': 'authCaptureTransaction',
+        }
+        if value_overrides:
+            default_transaction.update(value_overrides)
+
+        return TransactionData.create_new(default_transaction)
+
     @staticmethod
     def put_compact_active_member_jurisdictions(
         compact: str = DEFAULT_COMPACT, postal_abbreviations: list[str] = None
diff --git a/backend/compact-connect/lambdas/python/common/tests/__init__.py b/backend/compact-connect/lambdas/python/common/tests/__init__.py
index ee273a6f4..01c1fdd22 100644
--- a/backend/compact-connect/lambdas/python/common/tests/__init__.py
+++ b/backend/compact-connect/lambdas/python/common/tests/__init__.py
@@ -104,7 +104,9 @@ def setUpClass(cls):
         # Monkey-patch config object to be sure we have it based
         # on the env vars we set above
         import cc_common.config
+        from common_test.test_data_generator import TestDataGenerator
 
         cls.config = cc_common.config._Config()  # noqa: SLF001 protected-access
         cc_common.config.config = cls.config
         cls.mock_context = MagicMock(name='MockLambdaContext', spec=LambdaContext)
+        cls.test_data_generator = TestDataGenerator
diff --git a/backend/compact-connect/lambdas/python/common/tests/function/test_data_model/test_transaction_client.py b/backend/compact-connect/lambdas/python/common/tests/function/test_data_model/test_transaction_client.py
index 67f34b7d1..a525544dd 100644
--- a/backend/compact-connect/lambdas/python/common/tests/function/test_data_model/test_transaction_client.py
+++ b/backend/compact-connect/lambdas/python/common/tests/function/test_data_model/test_transaction_client.py
@@ -1,5 +1,5 @@
-import json
 from datetime import UTC, datetime, timedelta
+from unittest.mock import MagicMock, patch
 
 from moto import mock_aws
 
@@ -8,12 +8,19 @@
 
 @mock_aws
 class TestTransactionClient(TstFunction):
-    def _generate_mock_transaction(self, transaction_id: str, settlement_time_utc: str, batch_id: str) -> dict:
-        with open('tests/resources/dynamo/transaction.json') as f:
-            transaction = json.load(f)
-            transaction['transactionId'] = transaction_id
-            transaction['batch']['settlementTimeUTC'] = settlement_time_utc
-            transaction['batch']['batchId'] = batch_id
+    def _generate_mock_transaction(
+        self, transaction_id: str, compact: str = None, settlement_time_utc: str = None, batch_id: str = None
+    ):
+        transaction = self.test_data_generator.generate_default_transaction(
+            value_overrides={
+                'transactionId': transaction_id,
+                **({'compact': compact} if compact else {}),
+            }
+        )
+        if settlement_time_utc:
+            transaction.batch['settlementTimeUTC'] = settlement_time_utc
+        if batch_id:
+            transaction.batch['batchId'] = batch_id
         return transaction
 
     def test_transaction_history_edge_times(self):
@@ -34,12 +41,24 @@ def test_transaction_history_edge_times(self):
         client.store_transactions(
             transactions=[
                 # One at the beginning of the window
-                self._generate_mock_transaction(
-                    transaction_id='123', settlement_time_utc=start_time_string, batch_id='abc'
+                self.test_data_generator.generate_default_transaction(
+                    {
+                        'transactionId': '123',
+                        'batch': {
+                            'batchId': 'abc',
+                            'settlementTimeUTC': start_time_string,
+                        },
+                    }
                 ),
                 # One at the end of the window
-                self._generate_mock_transaction(
-                    transaction_id='456', settlement_time_utc=end_time_string, batch_id='def'
+                self.test_data_generator.generate_default_transaction(
+                    {
+                        'transactionId': '456',
+                        'batch': {
+                            'batchId': 'def',
+                            'settlementTimeUTC': end_time_string,
+                        },
+                    }
                 ),
             ],
         )
@@ -57,54 +76,19 @@ def test_transaction_history_edge_times(self):
         # remove dynamic dateOfUpdate timestamp
         transactions[0].pop('dateOfUpdate')
 
-        self.assertEqual(
+        expected_transaction = self.test_data_generator.generate_default_transaction(
             {
+                'transactionId': '123',
                 'batch': {
                     'batchId': 'abc',
-                    'settlementState': 'settledSuccessfully',
-                    'settlementTimeLocal': '2024-01-01T09:00:00',
                     'settlementTimeUTC': start_time_string,
                 },
-                'compact': 'aslp',
-                'licenseeId': '12345',
-                'lineItems': [
-                    {
-                        'description': 'Compact Privilege for Ohio',
-                        'itemId': 'priv:aslp-oh',
-                        'name': 'Ohio Compact Privilege',
-                        'privilegeId': 'mock-privilege-id-oh',
-                        'quantity': '1.0',
-                        'taxable': 'False',
-                        'unitPrice': '100.00',
-                    },
-                    {
-                        'description': 'Compact fee applied for each privilege purchased',
-                        'itemId': 'aslp-compact-fee',
-                        'name': 'ASLP Compact Fee',
-                        'quantity': '1',
-                        'taxable': 'False',
-                        'unitPrice': '10.50',
-                    },
-                    {
-                        'description': 'Credit card transaction fee',
-                        'itemId': 'credit-card-transaction-fee',
-                        'name': 'Credit Card Transaction Fee',
-                        'quantity': '1',
-                        'taxable': 'False',
-                        'unitPrice': '3.00',
-                    },
-                ],
-                'pk': 'COMPACT#aslp#TRANSACTIONS#MONTH#2024-01',
-                'responseCode': '1',
-                'settleAmount': '113.50',
-                'sk': 'COMPACT#aslp#TIME#1704067200#BATCH#abc#TX#123',
-                'submitTimeUTC': '2024-01-01T12:00:00.000Z',
-                'transactionId': '123',
-                'transactionProcessor': 'authorize.net',
-                'transactionStatus': 'settledSuccessfully',
-                'transactionType': 'authCaptureTransaction',
-                'type': 'transaction',
-            },
+            }
+        ).serialize_to_database_record()
+        expected_transaction.pop('dateOfUpdate')
+
+        self.assertEqual(
+            expected_transaction,
             transactions[0],
         )
 
@@ -186,8 +170,18 @@ def test_reconcile_unsettled_transactions_all_matched(self):
 
         # Create matching settled transactions
         settled_transactions = [
-            {'transactionId': 'tx-1', 'compact': compact},
-            {'transactionId': 'tx-2', 'compact': compact},
+            self.test_data_generator.generate_default_transaction(
+                {
+                    'transactionId': 'tx-1',
+                    'compact': compact,
+                }
+            ),
+            self.test_data_generator.generate_default_transaction(
+                {
+                    'transactionId': 'tx-2',
+                    'compact': compact,
+                }
+            ),
         ]
 
         client.reconcile_unsettled_transactions(compact=compact, settled_transactions=settled_transactions)
@@ -248,7 +242,7 @@ def test_reconcile_unsettled_transactions_deletes_matching_record_and_returns_ol
 
         # Only one settled transaction that matches
         settled_transactions = [
-            {'transactionId': 'tx-matched', 'compact': compact},
+            self.test_data_generator.generate_default_transaction({'transactionId': 'tx-matched', 'compact': compact}),
         ]
 
         result = client.reconcile_unsettled_transactions(compact=compact, settled_transactions=settled_transactions)
@@ -265,3 +259,176 @@ def test_reconcile_unsettled_transactions_deletes_matching_record_and_returns_ol
             ['tx-old-unmatched', 'tx-recent-unmatched'],
             [transaction['transactionId'] for transaction in response['Items']],
         )  # Two unmatched transactions remain
+
+    @patch('cc_common.data_model.transaction_client.logger')
+    def test_reconcile_unsettled_transactions_logs_error_when_settled_transactions_not_matched(
+        self, mock_logger
+    ):
+        """
+        Test that reconcile_unsettled_transactions logs an error when settled transactions don't match unsettled ones.
+        """
+        from cc_common.data_model.transaction_client import TransactionClient
+
+        client = TransactionClient(self.config)
+
+        # Create some unsettled transactions
+        compact = 'aslp'
+        transaction_date = datetime.now(UTC).isoformat()
+        client.store_unsettled_transaction(
+            compact=compact, transaction_id='tx-unsettled-1', transaction_date=transaction_date
+        )
+
+        # Create settled transactions - one matches, one doesn't have an unsettled record
+        settled_transactions = [
+            self.test_data_generator.generate_default_transaction(
+                {
+                    'transactionId': 'tx-unsettled-1',
+                    'compact': compact,
+                }
+            ),
+            self.test_data_generator.generate_default_transaction(
+                {
+                    'transactionId': 'tx-settled-without-unsettled',
+                    'compact': compact,
+                }
+            ),
+        ]
+
+        result = client.reconcile_unsettled_transactions(compact=compact, settled_transactions=settled_transactions)
+
+        # Verify logger.error was called with the expected message
+        mock_logger.error.assert_called_once()
+        call_args = mock_logger.error.call_args
+        self.assertEqual(
+            'Unable to reconcile some transactions from Authorize.Net with our unsettled transactions',
+            call_args[0][0],
+        )
+        self.assertEqual({'tx-settled-without-unsettled'}, call_args[1]['unreconciled_transactions'])
+
+        # Verify the method still returns correctly (old unsettled transactions)
+        self.assertEqual([], result)
+
+        # Verify matched unsettled transaction was deleted
+        pk = f'COMPACT#{compact}#UNSETTLED_TRANSACTIONS'
+        response = self._transaction_history_table.query(
+            KeyConditionExpression='pk = :pk', ExpressionAttributeValues={':pk': pk}
+        )
+        self.assertEqual(0, len(response['Items']))
+
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-11-15T12:00:00+00:00'))
+    def test_get_most_recent_transaction_for_compact_finds_in_current_month(self):
+        """Test that get_most_recent_transaction_for_compact finds the most recent transaction in the current month"""
+        from cc_common.data_model.transaction_client import TransactionClient
+
+        client = TransactionClient(self.config)
+
+        compact = 'aslp'
+
+        # Create transactions in the current month (November 2024)
+        # Format datetime as ISO string with Z suffix (e.g., '2024-11-01T00:00:00.000Z')
+        def format_utc_datetime(dt):
+            return dt.replace(tzinfo=None).isoformat() + '.000Z'
+
+        current_month_start = datetime(2024, 11, 1, tzinfo=UTC)
+        # Create two transactions - one older, one newer
+        older_transaction = self._generate_mock_transaction(
+            transaction_id='tx-older',
+            compact=compact,
+            settlement_time_utc=format_utc_datetime(current_month_start + timedelta(days=1)),
+            batch_id='batch-1',
+        )
+        newer_transaction = self._generate_mock_transaction(
+            transaction_id='tx-newer',
+            compact=compact,
+            settlement_time_utc=format_utc_datetime(current_month_start + timedelta(days=2)),
+            batch_id='batch-2',
+        )
+
+        # Store transactions
+        client.store_transactions(transactions=[older_transaction, newer_transaction])
+
+        # Get the most recent transaction
+        result = client.get_most_recent_transaction_for_compact(compact=compact)
+
+        # Should return the newer transaction
+        self.assertEqual('tx-newer', result.transactionId)
+
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-11-15T12:00:00+00:00'))
+    def test_get_most_recent_transaction_for_compact_searches_previous_months(self):
+        """Test that get_most_recent_transaction_for_compact searches previous months when current month is empty"""
+        from cc_common.data_model.transaction_client import TransactionClient
+
+        client = TransactionClient(self.config)
+
+        compact = 'aslp'
+
+        # Create a transaction in the previous month (October 2024)
+        # Format datetime as ISO string with Z suffix (e.g., '2024-10-15T00:00:00.000Z')
+        def format_utc_datetime(dt):
+            return dt.replace(tzinfo=None).isoformat() + '.000Z'
+
+        previous_month = datetime(2024, 10, 15, tzinfo=UTC)
+        previous_month_transaction = self._generate_mock_transaction(
+            transaction_id='tx-previous-month',
+            compact=compact,
+            settlement_time_utc=format_utc_datetime(previous_month),
+            batch_id='batch-prev',
+        )
+
+        # Store transaction
+        client.store_transactions(transactions=[previous_month_transaction])
+
+        # Get the most recent transaction
+        result = client.get_most_recent_transaction_for_compact(compact=compact)
+
+        # Should return the transaction from the previous month
+        self.assertEqual('tx-previous-month', result.transactionId)
+
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-02-01T01:00:00+00:00'))
+    def test_get_most_recent_transaction_for_compact_raises_when_no_transactions_found(self):
+        """Test that get_most_recent_transaction_for_compact raises ValueError when no transactions are found"""
+        from cc_common.data_model.transaction_client import TransactionClient
+
+        compact = 'aslp'
+
+        # Mock the query method to return empty results and track calls
+        # Create a mock table that returns empty results
+        mock_table = MagicMock()
+        mock_table.query.return_value = {'Items': []}
+
+        # Create a mock DynamoDB resource that returns our mock table
+        mock_resource = MagicMock()
+        mock_resource.Table.return_value = mock_table
+
+        # Patch boto3.resource to return our mock resource
+        with patch('cc_common.config.boto3.resource', return_value=mock_resource):
+            client = TransactionClient(self.config)
+
+            # Try to get the most recent transaction for a compact with no transactions
+            with self.assertRaises(ValueError) as context:
+                client.get_most_recent_transaction_for_compact(compact=compact)
+
+        # Verify ValueError is raised with correct message
+        self.assertIn('No transactions found for compact: aslp', str(context.exception))
+
+        # Verify query was called 3 times (once for each month)
+        self.assertEqual(3, mock_table.query.call_count)
+
+        # Verify each call queried the correct partition key
+        # First call: current month (2024-02)
+        first_call = mock_table.query.call_args_list[0]
+        # Key condition expressions provided to the Table.query call aren't very conducive to testing - we have
+        # to dig the values out
+        # They look like tuple(Key('pk'), '<private-key-value>')
+        first_condition_values = first_call.kwargs['KeyConditionExpression']._values  # noqa: SLF001
+        self.assertEqual(f'COMPACT#{compact}#TRANSACTIONS#MONTH#2024-02', first_condition_values[1])
+
+        # Second call: previous month (2024-01)
+        second_call = mock_table.query.call_args_list[1]
+        second_condition_values = second_call.kwargs['KeyConditionExpression']._values  # noqa: SLF001
+        self.assertIn(f'COMPACT#{compact}#TRANSACTIONS#MONTH#2024-01', second_condition_values)
+
+        # Third call: month before that (2023-12)
+        third_call = mock_table.query.call_args_list[2]
+        third_condition_values = third_call.kwargs['KeyConditionExpression']._values  # noqa: SLF001
+        self.assertIn(f'COMPACT#{compact}#TRANSACTIONS#MONTH#2023-12', third_condition_values)
diff --git a/backend/compact-connect/lambdas/python/common/tests/unit/test_data_model/test_schema/test_transaction.py b/backend/compact-connect/lambdas/python/common/tests/unit/test_data_model/test_schema/test_transaction.py
new file mode 100644
index 000000000..59228a81d
--- /dev/null
+++ b/backend/compact-connect/lambdas/python/common/tests/unit/test_data_model/test_schema/test_transaction.py
@@ -0,0 +1,128 @@
+from marshmallow import ValidationError
+
+from tests import TstLambdas
+
+
+class TestTransactionRecordSchema(TstLambdas):
+    def setUp(self):
+        from common_test.test_data_generator import TestDataGenerator
+
+        self.test_data_generator = TestDataGenerator
+
+    def test_serde(self):
+        """Test round-trip deserialization/serialization"""
+        from cc_common.data_model.schema.transaction.record import TransactionRecordSchema
+
+        expected_transaction = self.test_data_generator.generate_default_transaction().serialize_to_database_record()
+
+        schema = TransactionRecordSchema()
+        loaded_schema = schema.load(expected_transaction.copy())
+
+        transaction_data = schema.dump(loaded_schema)
+
+        # Drop dynamic fields
+        del expected_transaction['dateOfUpdate']
+        del transaction_data['dateOfUpdate']
+
+        self.assertEqual(expected_transaction, transaction_data)
+
+    def test_invalid(self):
+        from cc_common.data_model.schema.transaction.record import TransactionRecordSchema
+
+        transaction_data = self.test_data_generator.generate_default_transaction().to_dict()
+        transaction_data.pop('transactionId')
+
+        with self.assertRaises(ValidationError):
+            TransactionRecordSchema().load(transaction_data)
+
+    def test_invalid_transaction_processor(self):
+        from cc_common.data_model.schema.transaction import TransactionData
+
+        transaction_data = self.test_data_generator.generate_default_transaction()
+        transaction_record = transaction_data.serialize_to_database_record()
+        transaction_record['transactionProcessor'] = 'invalid-processor'
+
+        with self.assertRaises(ValidationError):
+            TransactionData.from_database_record(transaction_record)
+
+
+class TestTransactionDataClass(TstLambdas):
+    def setUp(self):
+        from common_test.test_data_generator import TestDataGenerator
+
+        self.test_data_generator = TestDataGenerator
+
+    def test_transaction_data_class_getters_return_expected_values(self):
+        from cc_common.data_model.schema.transaction import TransactionData
+
+        transaction_data = self.test_data_generator.generate_default_transaction().serialize_to_database_record()
+
+        transaction = TransactionData.from_database_record(transaction_data)
+        self.assertEqual(transaction.transactionProcessor, transaction_data['transactionProcessor'])
+        self.assertEqual(transaction.transactionId, transaction_data['transactionId'])
+        self.assertEqual(transaction.batch, transaction_data['batch'])
+        self.assertEqual(transaction.lineItems, transaction_data['lineItems'])
+        self.assertEqual(transaction.compact, transaction_data['compact'])
+        self.assertEqual(transaction.licenseeId, transaction_data['licenseeId'])
+        self.assertEqual(transaction.responseCode, transaction_data['responseCode'])
+        self.assertEqual(transaction.settleAmount, transaction_data['settleAmount'])
+        self.assertEqual(transaction.submitTimeUTC, transaction_data['submitTimeUTC'])
+        self.assertEqual(transaction.transactionStatus, transaction_data['transactionStatus'])
+        self.assertEqual(transaction.transactionType, transaction_data['transactionType'])
+
+    def test_transaction_data_class_outputs_expected_database_object(self):
+        # check final snapshot of expected data
+        transaction_data = self.test_data_generator.generate_default_transaction().serialize_to_database_record()
+        # remove dynamic field
+        del transaction_data['dateOfUpdate']
+
+        self.assertEqual(
+            {
+                'batch': {
+                    'batchId': '67890',
+                    'settlementState': 'settledSuccessfully',
+                    'settlementTimeLocal': '2024-01-01T09:00:00',
+                    'settlementTimeUTC': '2024-01-01T13:00:00.000Z',
+                },
+                'compact': 'aslp',
+                'licenseeId': '89a6377e-c3a5-40e5-bca5-317ec854c570',
+                'lineItems': [
+                    {
+                        'description': 'Compact Privilege for Ohio',
+                        'itemId': 'priv:aslp-oh',
+                        'name': 'Ohio Compact Privilege',
+                        'quantity': '1.0',
+                        'taxable': 'False',
+                        'unitPrice': '100.00',
+                        'privilegeId': 'mock-privilege-id-oh',
+                    },
+                    {
+                        'description': 'Compact fee applied for each privilege purchased',
+                        'itemId': 'aslp-compact-fee',
+                        'name': 'ASLP Compact Fee',
+                        'quantity': '1',
+                        'taxable': 'False',
+                        'unitPrice': '10.50',
+                    },
+                    {
+                        'description': 'credit card transaction fee',
+                        'itemId': 'credit-card-transaction-fee',
+                        'name': 'Credit Card Transaction Fee',
+                        'quantity': '1',
+                        'taxable': 'False',
+                        'unitPrice': '3.00',
+                    },
+                ],
+                'pk': 'COMPACT#aslp#TRANSACTIONS#MONTH#2024-01',
+                'responseCode': '1',
+                'settleAmount': '113.50',
+                'sk': 'COMPACT#aslp#TIME#1704114000#BATCH#67890#TX#1234567890',
+                'submitTimeUTC': '2024-01-01T12:00:00.000Z',
+                'transactionId': '1234567890',
+                'transactionProcessor': 'authorize.net',
+                'transactionStatus': 'settledSuccessfully',
+                'transactionType': 'authCaptureTransaction',
+                'type': 'transaction',
+            },
+            transaction_data,
+        )
diff --git a/backend/compact-connect/lambdas/python/common/tests/unit/test_data_model/test_transaction_client.py b/backend/compact-connect/lambdas/python/common/tests/unit/test_data_model/test_transaction_client.py
index 9dd7284d0..c207c69a6 100644
--- a/backend/compact-connect/lambdas/python/common/tests/unit/test_data_model/test_transaction_client.py
+++ b/backend/compact-connect/lambdas/python/common/tests/unit/test_data_model/test_transaction_client.py
@@ -1,8 +1,12 @@
-import json
+from copy import deepcopy
 from datetime import datetime
 from unittest.mock import ANY, MagicMock
 
 from boto3.dynamodb.conditions import Key
+from common_test.test_constants import (
+    DEFAULT_COMPACT_TRANSACTION_FEE_LINE_ITEM,
+    DEFAULT_COMPACT_TRANSACTION_PRIVILEGE_LINE_ITEM,
+)
 
 from tests import TstLambdas
 
@@ -10,14 +14,6 @@
 
 
 class TestTransactionClient(TstLambdas):
-    def _generate_mock_transaction(self, transaction_id: str, settlement_time_utc: str, batch_id: str) -> dict:
-        with open('tests/resources/dynamo/transaction.json') as f:
-            transaction = json.load(f)
-            transaction['transactionId'] = transaction_id
-            transaction['batch']['settlementTimeUTC'] = settlement_time_utc
-            transaction['batch']['batchId'] = batch_id
-        return transaction
-
     def setUp(self):
         from cc_common.data_model import transaction_client
 
@@ -31,8 +27,14 @@ def setUp(self):
         self.client = transaction_client.TransactionClient(self.mock_config)
 
     def test_store_transactions_authorize_net(self):
-        mock_transaction = self._generate_mock_transaction(
-            transaction_id='tx123', settlement_time_utc=TEST_SETTLEMENT_DATETIME, batch_id='batch456'
+        mock_transaction = self.test_data_generator.generate_default_transaction(
+            {
+                'transactionId': 'tx123',
+                'batch': {
+                    'batchId': 'batch456',
+                    'settlementTimeUTC': TEST_SETTLEMENT_DATETIME,
+                },
+            }
         )
         # Test data
         test_transactions = [
@@ -44,7 +46,7 @@ def test_store_transactions_authorize_net(self):
 
         # Verify the batch writer was called with correct data
         expected_epoch = int(datetime.fromisoformat(TEST_SETTLEMENT_DATETIME).timestamp())
-        expected_item = mock_transaction.copy()
+        expected_item = mock_transaction.to_dict().copy()
         expected_item.update(
             {
                 'pk': 'COMPACT#aslp#TRANSACTIONS#MONTH#2024-01',
@@ -59,27 +61,19 @@ def test_store_transactions_authorize_net(self):
 
     def test_store_transactions_unsupported_processor(self):
         # Test data with unsupported processor
-        test_transactions = [
-            {
-                'transactionProcessor': 'unsupported',
-                'transactionId': 'tx123',
-                'batch': {'batchId': 'batch456', 'settlementTimeUTC': '2024-01-15T10:30:00+00:00'},
-            }
-        ]
+        transaction = self.test_data_generator.generate_default_transaction()
+        # We'll force the transaction into an invalid state by updating the internal data, after it's done validation
+        transaction._data['transactionProcessor'] = 'unsupported'  # noqa: SLF001
 
         # Verify it raises ValueError for unsupported processor
         with self.assertRaises(ValueError):
-            self.client.store_transactions(test_transactions)
+            self.client.store_transactions([transaction])
 
     def test_store_multiple_transactions(self):
         # Test data with multiple transactions
         test_transactions = [
-            self._generate_mock_transaction(
-                transaction_id='tx123', settlement_time_utc=TEST_SETTLEMENT_DATETIME, batch_id='batch456'
-            ),
-            self._generate_mock_transaction(
-                transaction_id='tx124', settlement_time_utc=TEST_SETTLEMENT_DATETIME, batch_id='batch456'
-            ),
+            self.test_data_generator.generate_default_transaction({'transactionId': 'tx123'}),
+            self.test_data_generator.generate_default_transaction({'transactionId': 'tx124'}),
         ]
 
         # Call the method
@@ -110,16 +104,20 @@ def test_add_privilege_information_to_transactions(self):
         }
 
         # Test data
+        priv_line_ca = deepcopy(DEFAULT_COMPACT_TRANSACTION_PRIVILEGE_LINE_ITEM)
+        priv_line_ca.update({'itemId': 'priv:aslp-CA', 'unitPrice': 100})
+        priv_line_ny = deepcopy(DEFAULT_COMPACT_TRANSACTION_PRIVILEGE_LINE_ITEM)
+        priv_line_ny.update({'itemId': 'priv:aslp-NY', 'unitPrice': 200})
+        priv_line_fee = deepcopy(DEFAULT_COMPACT_TRANSACTION_FEE_LINE_ITEM)
+        priv_line_fee.update({'itemId': 'credit-card-transaction-fee', 'unitPrice': 50})
         test_transactions = [
-            {
-                'transactionId': 'tx123',
-                'licenseeId': 'prov-123',
-                'lineItems': [
-                    {'itemId': 'priv:aslp-CA', 'unitPrice': 100},
-                    {'itemId': 'priv:aslp-NY', 'unitPrice': 200},
-                    {'itemId': 'credit-card-transaction-fee', 'unitPrice': 50},
-                ],
-            }
+            self.test_data_generator.generate_default_transaction(
+                {
+                    'transactionId': 'tx123',
+                    'licenseeId': 'prov-123',
+                    'lineItems': [priv_line_ca, priv_line_ny, priv_line_fee],
+                }
+            )
         ]
 
         # Call the method
@@ -132,9 +130,9 @@ def test_add_privilege_information_to_transactions(self):
         )
 
         # Verify privilege IDs were added to correct line items
-        self.assertEqual(result[0]['lineItems'][0]['privilegeId'], 'priv-123')  # CA line item
-        self.assertEqual(result[0]['lineItems'][1]['privilegeId'], 'priv-456')  # NY line item
-        self.assertNotIn('privilegeId', result[0]['lineItems'][2])  # other item
+        self.assertEqual(result[0].lineItems[0]['privilegeId'], 'priv-123')  # CA line item
+        self.assertEqual(result[0].lineItems[1]['privilegeId'], 'priv-456')  # NY line item
+        self.assertNotIn('privilegeId', result[0].lineItems[2])  # other item
 
     def test_add_privilege_information_to_transactions_maps_provider_id_to_transaction(self):
         expected_provider_id = 'abcd1234-5678-9012-3456-7890a0d12345'
@@ -154,16 +152,18 @@ def test_add_privilege_information_to_transactions_maps_provider_id_to_transacti
         }
 
         # Test data
+        line_item_ca = deepcopy(DEFAULT_COMPACT_TRANSACTION_PRIVILEGE_LINE_ITEM)
+        line_item_ca.update({'itemId': 'priv:aslp-CA', 'unitPrice': 100})
+        line_item_fee = deepcopy(DEFAULT_COMPACT_TRANSACTION_FEE_LINE_ITEM)
+        line_item_fee.update({'itemId': 'credit-card-transaction-fee', 'unitPrice': 50})
         test_transactions = [
-            {
-                'transactionId': 'tx123',
-                # reproducing real case where licensee id was masked in authorize.net
-                'licenseeId': 'abcdXXXXXXXXXXXXXXXXXXX-4927a0d12345',
-                'lineItems': [
-                    {'itemId': 'priv:aslp-CA', 'unitPrice': 100},
-                    {'itemId': 'credit-card-transaction-fee', 'unitPrice': 50},
-                ],
-            }
+            self.test_data_generator.generate_default_transaction(
+                {
+                    'transactionId': 'tx123',
+                    'licenseeId': 'abcdXXXXXXXXXXXXXXXXXXX-4927a0d12345',
+                    'lineItems': [line_item_ca, line_item_fee],
+                }
+            ),
         ]
 
         # Call the method
@@ -176,7 +176,7 @@ def test_add_privilege_information_to_transactions_maps_provider_id_to_transacti
         )
 
         # Verify the correct provider ID was added to the transaction
-        self.assertEqual(expected_provider_id, result[0]['licenseeId'])
+        self.assertEqual(expected_provider_id, result[0].licenseeId)
         # Verify the privilege id is mapped as expected
-        self.assertEqual(expected_privilege_id, result[0]['lineItems'][0]['privilegeId'])
-        self.assertNotIn('privilegeId', result[0]['lineItems'][1])  # credit card fee line item
+        self.assertEqual(expected_privilege_id, result[0].lineItems[0]['privilegeId'])
+        self.assertNotIn('privilegeId', result[0].lineItems[1])  # credit card fee line item
diff --git a/backend/compact-connect/lambdas/python/purchases/handlers/transaction_history.py b/backend/compact-connect/lambdas/python/purchases/handlers/transaction_history.py
index f51b3e7c1..6ac434b6e 100644
--- a/backend/compact-connect/lambdas/python/purchases/handlers/transaction_history.py
+++ b/backend/compact-connect/lambdas/python/purchases/handlers/transaction_history.py
@@ -41,141 +41,173 @@ def process_settled_transactions(event: dict, context: LambdaContext) -> dict:
     last_processed_transaction_id = event.get('lastProcessedTransactionId')
     current_batch_id = event.get('currentBatchId')
     processed_batch_ids = event.get('processedBatchIds', [])
+    with logger.append_context_keys(
+        compact=compact,
+        scheduled_time=scheduled_time,
+        last_processed_transaction_id=last_processed_transaction_id,
+        current_batch_id=current_batch_id,
+    ):
+        # Get compact configuration and jurisdictions that are live for licensee registration
+        compact_configuration_client = config.compact_configuration_client
+        compact_configuration_options = compact_configuration_client.get_privilege_purchase_options(compact=compact)
+
+        compact_configuration = next(
+            (Compact(item) for item in compact_configuration_options['items'] if item['type'] == COMPACT_TYPE), None
+        )
+        jurisdiction_configurations = [
+            item for item in compact_configuration_options['items'] if item['type'] == JURISDICTION_TYPE
+        ]
+
+        if not compact_configuration or not jurisdiction_configurations:
+            logger.warning('The compact is not yet live - batch settlement data')
+            return {
+                'compact': compact,  # Always include the compact name
+                'scheduledTime': scheduled_time,  # Preserve scheduled time for subsequent iterations
+                'status': 'COMPLETE',  # This will skip straight to the end of the step function flow
+            }
+
+        start_time_str, end_time_str = _get_time_window_strings(event, compact)
 
-    # Get compact configuration and jurisdictions that are live for licensee registration
-    compact_configuration_client = config.compact_configuration_client
-    compact_configuration_options = compact_configuration_client.get_privilege_purchase_options(compact=compact)
+        logger.info(
+            'Collecting settled transaction for time period',
+            start_time=start_time_str,
+            end_time=end_time_str,
+        )
+
+        # Get transactions from payment processor
+        purchase_client = PurchaseClient()
+        transaction_response = purchase_client.get_settled_transactions(
+            compact=compact,
+            start_time=start_time_str,
+            end_time=end_time_str,
+            # we set the transaction limit to 500 to avoid hitting the 15-minute timeout for lambda
+            transaction_limit=500,
+            last_processed_transaction_id=last_processed_transaction_id,
+            current_batch_id=current_batch_id,
+            processed_batch_ids=processed_batch_ids,
+        )
 
-    compact_configuration = next(
-        (Compact(item) for item in compact_configuration_options['items'] if item['type'] == COMPACT_TYPE), None
-    )
-    jurisdiction_configurations = [
-        item for item in compact_configuration_options['items'] if item['type'] == JURISDICTION_TYPE
-    ]
+        # Store transactions in DynamoDB
+        if transaction_response['transactions']:
+            logger.info('Fetching privilege ids for transactions', compact=compact)
+            # first we must add the associated privilege ids to each transaction so we can show the association in our
+            # reports
+            transactions_with_privilege_ids = config.transaction_client.add_privilege_information_to_transactions(
+                compact=compact, transactions=transaction_response['transactions']
+            )
+            logger.info('Storing transactions in DynamoDB', compact=compact)
+            config.transaction_client.store_transactions(transactions=transactions_with_privilege_ids)
 
-    if not compact_configuration or not jurisdiction_configurations:
-        logger.warning('The compact is not yet live - batch settlement data')
-        return {
+        # Return appropriate response based on whether there are more transactions to process
+        response = {
             'compact': compact,  # Always include the compact name
             'scheduledTime': scheduled_time,  # Preserve scheduled time for subsequent iterations
-            'status': 'COMPLETE',  # This will skip straight to the end of the step function flow
+            'startTime': start_time_str,
+            'endTime': end_time_str,
+            'status': 'IN_PROGRESS' if not _all_transactions_processed(transaction_response) else 'COMPLETE',
+            'processedBatchIds': transaction_response['processedBatchIds'],
         }
 
-    # Use the scheduled time from EventBridge for replay-ability
-    # By default, the authorize.net accounts batch settlements at 4:00pm Pacific Time.
-    # This daily collector runs an hour later (5pm PST, which is 1am UTC) to collect
-    # all settled transaction for the last 24 hours.
-    end_time = datetime.fromisoformat(scheduled_time).replace(hour=1, minute=0, second=0, microsecond=0)
-    start_time = end_time - timedelta(days=1)
-
-    # Format timestamps for API call
-    start_time_str = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
-    end_time_str = end_time.strftime('%Y-%m-%dT%H:%M:%SZ')
-
-    logger.info(
-        'Collecting settled transaction for time period',
-        compact=compact,
-        start_time=start_time_str,
-        end_time=end_time_str,
-    )
+        # Only include pagination values if we're not done processing
+        if not _all_transactions_processed(transaction_response):
+            logger.info('Not all transactions processed, updating response with pagination values')
+            response.update(
+                {
+                    'lastProcessedTransactionId': transaction_response['lastProcessedTransactionId'],
+                    'currentBatchId': transaction_response['currentBatchId'],
+                }
+            )
 
-    # Get transactions from payment processor
-    purchase_client = PurchaseClient()
-    transaction_response = purchase_client.get_settled_transactions(
-        compact=compact,
-        start_time=start_time_str,
-        end_time=end_time_str,
-        # we set the transaction limit to 500 to avoid hitting the 15-minute timeout for lambda
-        transaction_limit=500,
-        last_processed_transaction_id=last_processed_transaction_id,
-        current_batch_id=current_batch_id,
-        processed_batch_ids=processed_batch_ids,
-    )
-
-    # Store transactions in DynamoDB
-    if transaction_response['transactions']:
-        logger.info('Fetching privilege ids for transactions', compact=compact)
-        # first we must add the associated privilege ids to each transaction so we can show the association in our
-        # reports
-        transactions_with_privilege_ids = config.transaction_client.add_privilege_information_to_transactions(
-            compact=compact, transactions=transaction_response['transactions']
-        )
-        logger.info('Storing transactions in DynamoDB', compact=compact)
-        config.transaction_client.store_transactions(transactions=transactions_with_privilege_ids)
-
-    # Return appropriate response based on whether there are more transactions to process
-    response = {
-        'compact': compact,  # Always include the compact name
-        'scheduledTime': scheduled_time,  # Preserve scheduled time for subsequent iterations
-        'status': 'IN_PROGRESS' if not _all_transactions_processed(transaction_response) else 'COMPLETE',
-        'processedBatchIds': transaction_response['processedBatchIds'],
-    }
-
-    # Only include pagination values if we're not done processing
-    if not _all_transactions_processed(transaction_response):
-        logger.info('Not all transactions processed, updating response with pagination values')
-        response.update(
-            {
-                'lastProcessedTransactionId': transaction_response['lastProcessedTransactionId'],
-                'currentBatchId': transaction_response['currentBatchId'],
-            }
+        # here we check if there were any settlement errors in the batch or if there was a settlement failure
+        # in a previous iteration, and we need to send an alert to the compact operations team
+        failed_transactions_ids = transaction_response.get('settlementErrorTransactionIds', [])
+        if failed_transactions_ids or event.get('batchFailureErrorMessage'):
+            # error message should be a json object we can load
+            if event.get('batchFailureErrorMessage'):
+                batch_failure_error_message = json.loads(event.get('batchFailureErrorMessage'))
+                batch_failure_error_message['failedTransactionIds'].extend(failed_transactions_ids)
+                response['batchFailureErrorMessage'] = json.dumps(batch_failure_error_message)
+            else:
+                # if there was no previous error message, we'll create a new one
+                response['batchFailureErrorMessage'] = json.dumps(
+                    {
+                        'message': 'Settlement errors detected in one or more transactions.',
+                        'failedTransactionIds': failed_transactions_ids,
+                    }
+                )
+
+            if _all_transactions_processed(transaction_response):
+                # we've finished storing all transactions for this period,
+                # and we need to send an alert to the compact operations team
+                logger.error(
+                    'Batch settlement error detected', batchFailureErrorMessage=response['batchFailureErrorMessage']
+                )
+                response['status'] = 'BATCH_FAILURE'
+
+        # Reconcile unsettled transactions with settled transactions
+        # This must be run every iteration to clean up all unsettled transaction records from the transaction history
+        # table
+        old_unsettled_transaction_ids = config.transaction_client.reconcile_unsettled_transactions(
+            compact=compact, settled_transactions=transaction_response['transactions']
         )
 
-    # here we check if there were any settlement errors in the batch or if there was a settlement failure
-    # in a previous iteration, and we need to send an alert to the compact operations team
-    failed_transactions_ids = transaction_response.get('settlementErrorTransactionIds', [])
-    if failed_transactions_ids or event.get('batchFailureErrorMessage'):
-        # error message should be a json object we can load
-        if event.get('batchFailureErrorMessage'):
-            batch_failure_error_message = json.loads(event.get('batchFailureErrorMessage'))
-            batch_failure_error_message['failedTransactionIds'].extend(failed_transactions_ids)
-            response['batchFailureErrorMessage'] = json.dumps(batch_failure_error_message)
-        else:
-            # if there was no previous error message, we'll create a new one
-            response['batchFailureErrorMessage'] = json.dumps(
+        # Check for old unsettled transactions (older than 48 hours) and report on them if we've
+        # finished processing all transactions
+        if old_unsettled_transaction_ids and _all_transactions_processed(transaction_response):
+            # Parse existing error message if it exists
+            existing_error = {}
+            if response.get('batchFailureErrorMessage'):
+                existing_error = json.loads(response.get('batchFailureErrorMessage'))
+
+            existing_error.update(
                 {
-                    'message': 'Settlement errors detected in one or more transactions.',
-                    'failedTransactionIds': failed_transactions_ids,
+                    'message': existing_error.get('message', '')
+                    + ' One or more transactions have not settled in over 48 hours.',
+                    'unsettledTransactionIds': old_unsettled_transaction_ids,
                 }
             )
 
-        if _all_transactions_processed(transaction_response):
-            # we've finished storing all transactions for this period,
-            # and we need to send an alert to the compact operations team
+            response['batchFailureErrorMessage'] = json.dumps(existing_error)
+
             logger.error(
-                'Batch settlement error detected', batchFailureErrorMessage=response['batchFailureErrorMessage']
+                'Unsettled transactions older than 48 hours detected',
+                unsettledTransactionIds=old_unsettled_transaction_ids,
             )
             response['status'] = 'BATCH_FAILURE'
 
-    # Reconcile unsettled transactions with settled transactions
-    # This must be run every iteration to clean up all unsettled transaction records from the transaction history
-    # table
-    old_unsettled_transaction_ids = config.transaction_client.reconcile_unsettled_transactions(
-        compact=compact, settled_transactions=transaction_response['transactions']
-    )
-
-    # Check for old unsettled transactions (older than 48 hours) and report on them if we've
-    # finished processing all transactions
-    if old_unsettled_transaction_ids and _all_transactions_processed(transaction_response):
-        # Parse existing error message if it exists
-        existing_error = {}
-        if response.get('batchFailureErrorMessage'):
-            existing_error = json.loads(response.get('batchFailureErrorMessage'))
-
-        existing_error.update(
-            {
-                'message': existing_error.get('message', '')
-                + ' One or more transactions have not settled in over 48 hours.',
-                'unsettledTransactionIds': old_unsettled_transaction_ids,
-            }
-        )
+        return response
 
-        response['batchFailureErrorMessage'] = json.dumps(existing_error)
 
-        logger.error(
-            'Unsettled transactions older than 48 hours detected',
-            unsettledTransactionIds=old_unsettled_transaction_ids,
+def _get_time_window_strings(event: dict, compact: str):
+    if event.get('startTime') and event.get('endTime'):
+        start_time = datetime.fromisoformat(event['startTime'])
+        end_time = datetime.fromisoformat(event['endTime'])
+        logger.info(
+            'Using start/end times from event', start_time=start_time.isoformat(), end_time=end_time.isoformat()
         )
-        response['status'] = 'BATCH_FAILURE'
+    else:
+        logger.info('Calculating start/end times')
+        # We collect a window spanning the most recent captured batch of settled transactions, through the scheduled
+        # EventBridge time. Nominally this is a 1-day window, but if something goes wrong and causes a batch to be
+        # delayed the window will expand in the subsequent run to pick up where it left off.
+        end_time = datetime.fromisoformat(event['scheduledTime']).replace(hour=1, minute=0, second=0, microsecond=0)
+        # Authorize.net will only allow us to query a 31-day window, so we'll limit ourselves to 30. If we fail to
+        # collect settled transactions for 30 days, we're well outside normal operations and will require manual
+        # intervention to recover data
+        oldest_allowed_start = end_time - timedelta(days=30)
+        try:
+            most_recent_settled_transaction = config.transaction_client.get_most_recent_transaction_for_compact(compact)
+            # Time ranges are inclusive in the Authorize.net API, so we need to shift our start forward by 1 second
+            most_recent_settlement = datetime.fromisoformat(
+                most_recent_settled_transaction.batch['settlementTimeUTC']
+            ) + timedelta(seconds=1)
+        except ValueError as e:
+            # We should make some noise if we can't find any transactions, but it's also an expected state for a compact
+            # that just went live, so we do need to be able to collect our first batch after launch. If we're in this
+            # state we'll log an error and collect what we can.
+            logger.warning('Failed to find transactions for compact', exc_info=e)
+            most_recent_settlement = oldest_allowed_start
+        start_time = max(most_recent_settlement, oldest_allowed_start)
 
-    return response
+    # Format timestamps for API call
+    return start_time.strftime('%Y-%m-%dT%H:%M:%SZ'), end_time.strftime('%Y-%m-%dT%H:%M:%SZ')
diff --git a/backend/compact-connect/lambdas/python/purchases/purchase_client.py b/backend/compact-connect/lambdas/python/purchases/purchase_client.py
index b668acd59..18c8168fb 100644
--- a/backend/compact-connect/lambdas/python/purchases/purchase_client.py
+++ b/backend/compact-connect/lambdas/python/purchases/purchase_client.py
@@ -19,6 +19,7 @@
 from cc_common.data_model.schema.compact import Compact
 from cc_common.data_model.schema.compact.common import CompactFeeType, PaymentProcessorType, TransactionFeeChargeType
 from cc_common.data_model.schema.jurisdiction import Jurisdiction
+from cc_common.data_model.schema.transaction import TransactionData
 from cc_common.exceptions import (
     CCFailedTransactionException,
     CCInternalException,
@@ -39,6 +40,10 @@
 AUTHORIZE_NET_CARD_USER_ERROR_CODES = ['2', '5', '6', '7', '8', '11', '17', '65', 'E00114']
 
 
+class AuthorizeNetTransactionIgnoreStates(StrEnum):
+    DeclinedError = 'declined'
+
+
 class AuthorizeNetTransactionErrorStates(StrEnum):
     SettlementError = 'settlementError'
     GeneralError = 'generalError'
@@ -218,6 +223,7 @@ def validate_credentials(self) -> dict:
     @abstractmethod
     def get_settled_transactions(
         self,
+        compact: str,
         start_time: str,
         end_time: str,
         transaction_limit: int,
@@ -228,6 +234,7 @@ def get_settled_transactions(
         """
         Get settled transactions from the payment processor.
 
+        :param compact: The compact abbreviation
         :param start_time: UTC timestamp string for start of range
         :param end_time: UTC timestamp string for end of range
         :param transaction_limit: Maximum number of transactions to return
@@ -727,6 +734,7 @@ def _get_transaction_details(self, transaction_id: str) -> apicontractsv1.getTra
 
     def get_settled_transactions(
         self,
+        compact: str,
         start_time: str,
         end_time: str,
         transaction_limit: int,
@@ -737,6 +745,7 @@ def get_settled_transactions(
         """
         Get settled transactions from the payment processor.
 
+        :param compact: The compact abbreviation
         :param start_time: UTC timestamp string for start of range
         :param end_time: UTC timestamp string for end of range
         :param transaction_limit: Maximum number of transactions to return
@@ -815,6 +824,14 @@ def get_settled_transactions(
                                     transaction_id=str(transaction.transId),
                                     transaction_status=str(tx.transactionStatus),
                                 )
+                            if str(tx.transactionStatus) in AuthorizeNetTransactionIgnoreStates:
+                                logger.info(
+                                    'Transaction was in an ignorable state. Skipping.',
+                                    batch_id=batch_id,
+                                    transaction_id=str(tx.transId),
+                                    transaction_status=str(tx.transactionStatus),
+                                )
+                                continue
 
                             licensee_id = None
                             if hasattr(tx, 'order') and tx.order.description:
@@ -839,26 +856,29 @@ def get_settled_transactions(
                                         }
                                     )
 
-                            transaction_data = {
-                                # we must cast these to strings, or they will cause an error when we try to serialize
-                                # in other parts of the system
-                                'transactionId': str(tx.transId),
-                                'submitTimeUTC': str(tx.submitTimeUTC),
-                                'transactionType': str(tx.transactionType),
-                                'transactionStatus': str(tx.transactionStatus),
-                                'responseCode': str(tx.responseCode),
-                                'settleAmount': str(tx.settleAmount),
-                                'licenseeId': licensee_id,
-                                'batch': {
-                                    'batchId': str(batch.batchId),
-                                    'settlementTimeUTC': str(batch.settlementTimeUTC),
-                                    'settlementTimeLocal': str(batch.settlementTimeLocal),
-                                    'settlementState': str(batch.settlementState),
-                                },
-                                'lineItems': line_items,
-                                # this defines the type of transaction processor that processed the transaction
-                                'transactionProcessor': PaymentProcessorType.AUTHORIZE_DOT_NET_TYPE,
-                            }
+                            transaction_data = TransactionData.create_new(
+                                {
+                                    # we must cast these to strings, or they will cause an error when we try to
+                                    # serialize in other parts of the system
+                                    'transactionId': str(tx.transId),
+                                    'submitTimeUTC': str(tx.submitTimeUTC),
+                                    'transactionType': str(tx.transactionType),
+                                    'transactionStatus': str(tx.transactionStatus),
+                                    'responseCode': str(tx.responseCode),
+                                    'settleAmount': str(tx.settleAmount),
+                                    'licenseeId': licensee_id,
+                                    'batch': {
+                                        'batchId': str(batch.batchId),
+                                        'settlementTimeUTC': str(batch.settlementTimeUTC),
+                                        'settlementTimeLocal': str(batch.settlementTimeLocal),
+                                        'settlementState': str(batch.settlementState),
+                                    },
+                                    'lineItems': line_items,
+                                    # this defines the type of transaction processor that processed the transaction
+                                    'transactionProcessor': PaymentProcessorType.AUTHORIZE_DOT_NET_TYPE,
+                                    'compact': compact,
+                                }
+                            )
                             transactions.append(transaction_data)
                             processed_transaction_count += 1
                             if processed_transaction_count >= transaction_limit:
@@ -1084,7 +1104,8 @@ def get_settled_transactions(
         if not self.payment_processor_client:
             self.payment_processor_client = self._get_compact_payment_processor_client(compact)
 
-        response = self.payment_processor_client.get_settled_transactions(
+        return self.payment_processor_client.get_settled_transactions(
+            compact=compact,
             start_time=start_time,
             end_time=end_time,
             transaction_limit=transaction_limit,
@@ -1092,9 +1113,3 @@ def get_settled_transactions(
             current_batch_id=current_batch_id,
             processed_batch_ids=processed_batch_ids,
         )
-
-        # Add compact to each transaction for serialization
-        for transaction in response['transactions']:
-            transaction['compact'] = compact
-
-        return response
diff --git a/backend/compact-connect/lambdas/python/purchases/tests/function/test_handlers/test_transaction_history.py b/backend/compact-connect/lambdas/python/purchases/tests/function/test_handlers/test_transaction_history.py
index 3e77d9a7e..de9816eda 100644
--- a/backend/compact-connect/lambdas/python/purchases/tests/function/test_handlers/test_transaction_history.py
+++ b/backend/compact-connect/lambdas/python/purchases/tests/function/test_handlers/test_transaction_history.py
@@ -1,6 +1,5 @@
 import json
 from datetime import UTC, datetime, timedelta
-from decimal import Decimal
 from unittest.mock import ANY, MagicMock, patch
 
 from moto import mock_aws
@@ -29,7 +28,7 @@
 MOCK_SCHEDULED_TIME = '2024-01-01T01:00:00Z'
 
 # Test jurisdiction data
-OHIO_JURISDICTION = {'postalAbbreviation': 'oh', 'jurisdictionName': 'ohio', 'sk': 'aslp#JURISDICTION#oh'}
+OHIO_JURISDICTION = {'postalAbbreviation': 'oh', 'jurisdictionName': 'ohio'}
 
 
 def _generate_mock_transaction(
@@ -38,30 +37,14 @@ def _generate_mock_transaction(
     transaction_status='settledSuccessfully',
     batch_settlement_state='settledSuccessfully',
 ):
+    from common_test.test_data_generator import TestDataGenerator
+
     if jurisdictions is None:
         jurisdictions = ['oh']
 
-    transaction = {
-        'transactionId': transaction_id,
-        'submitTimeUTC': MOCK_SUBMIT_TIME_UTC,
-        'transactionType': 'authCaptureTransaction',
-        'transactionStatus': transaction_status,
-        'responseCode': '1',
-        'settleAmount': '100.00',
-        'licenseeId': MOCK_LICENSEE_ID,
-        'batch': {
-            'batchId': MOCK_BATCH_ID,
-            'settlementTimeUTC': MOCK_SETTLEMENT_TIME_UTC,
-            'settlementTimeLocal': MOCK_SETTLEMENT_TIME_LOCAL,
-            'settlementState': batch_settlement_state,
-        },
-        'lineItems': [],
-        'compact': TEST_COMPACT,
-        'transactionProcessor': 'authorize.net',
-    }
-
+    line_items = []
     for jurisdiction in jurisdictions:
-        transaction['lineItems'].append(
+        line_items.append(
             {
                 'itemId': f'priv:{TEST_COMPACT}-{jurisdiction}-{TEST_AUD_LICENSE_TYPE_ABBR}',
                 'name': f'{jurisdiction.upper()} Compact Privilege',
@@ -72,7 +55,23 @@ def _generate_mock_transaction(
             }
         )
 
-    return transaction
+    return TestDataGenerator.generate_default_transaction(
+        {
+            'transactionId': transaction_id,
+            'submitTimeUTC': MOCK_SUBMIT_TIME_UTC,
+            'transactionStatus': transaction_status,
+            'settleAmount': '100.00',
+            'licenseeId': MOCK_LICENSEE_ID,
+            'batch': {
+                'batchId': MOCK_BATCH_ID,
+                'settlementTimeUTC': MOCK_SETTLEMENT_TIME_UTC,
+                'settlementTimeLocal': MOCK_SETTLEMENT_TIME_LOCAL,
+                'settlementState': batch_settlement_state,
+            },
+            'lineItems': line_items,
+            'compact': TEST_COMPACT,
+        }
+    )
 
 
 @mock_aws
@@ -81,22 +80,31 @@ class TestProcessSettledTransactions(TstFunction):
 
     def _add_compact_configuration_data(self, jurisdictions=None):
         """
-        Use the canned test resources to load compact and jurisdiction information into the DB.
+        Use the test data generator to load compact and jurisdiction information into the DB.
 
         If jurisdictions is None, it will default to only include Ohio.
         """
+        # Add jurisdiction configurations first
         if jurisdictions is None:
             jurisdictions = [OHIO_JURISDICTION]
 
-        with open('../common/tests/resources/dynamo/compact.json') as f:
-            record = json.load(f, parse_float=Decimal)
-            self._compact_configuration_table.put_item(Item=record)
+        for jurisdiction in jurisdictions:
+            self.test_data_generator.put_default_jurisdiction_configuration_in_configuration_table(
+                {
+                    'compact': TEST_COMPACT,
+                    'postalAbbreviation': jurisdiction['postalAbbreviation'],
+                    'jurisdictionName': jurisdiction['jurisdictionName'],
+                }
+            )
 
-        with open('../common/tests/resources/dynamo/jurisdiction.json') as f:
-            record = json.load(f, parse_float=Decimal)
-            for jurisdiction in jurisdictions:
-                record.update(jurisdiction)
-                self._compact_configuration_table.put_item(Item=record)
+        # Add compact configuration with configuredStates set based on jurisdictions
+        configured_states = [{'postalAbbreviation': j['postalAbbreviation'], 'isLive': True} for j in jurisdictions]
+        self.test_data_generator.put_default_compact_configuration_in_configuration_table(
+            {
+                'compactAbbr': TEST_COMPACT,
+                'configuredStates': configured_states,
+            }
+        )
 
     def _add_mock_privilege_to_database(
         self,
@@ -105,25 +113,15 @@ def _add_mock_privilege_to_database(
         transaction_id=MOCK_TRANSACTION_ID,
         jurisdiction='oh',
     ):
-        from cc_common.data_model.schema.privilege.record import PrivilegeRecordSchema
-
-        privilege_schema = PrivilegeRecordSchema()
-
-        with open('../common/tests/resources/dynamo/privilege.json') as f:
-            record = json.load(f)
-            loaded_record = privilege_schema.load(record)
-            loaded_record.update(
-                {
-                    'privilegeId': privilege_id,
-                    'providerId': licensee_id,
-                    'compact': TEST_COMPACT,
-                    'jurisdiction': jurisdiction,
-                    'compactTransactionId': transaction_id,
-                }
-            )
-
-            serialized_record = privilege_schema.dump(loaded_record)
-            self._provider_table.put_item(Item=serialized_record)
+        self.test_data_generator.put_default_privilege_record_in_provider_table(
+            {
+                'privilegeId': privilege_id,
+                'providerId': licensee_id,
+                'compact': TEST_COMPACT,
+                'jurisdiction': jurisdiction,
+                'compactTransactionId': transaction_id,
+            }
+        )
 
     def _add_mock_privilege_update_to_database(
         self,
@@ -132,31 +130,30 @@ def _add_mock_privilege_update_to_database(
         transaction_id=MOCK_TRANSACTION_ID,
         jurisdiction='oh',
     ):
-        from cc_common.data_model.schema.privilege.record import PrivilegeUpdateRecordSchema
-
-        privilege_update_schema = PrivilegeUpdateRecordSchema()
-
-        with open('../common/tests/resources/dynamo/privilege-update.json') as f:
-            record = json.load(f)
-            loaded_record = privilege_update_schema.load(record)
-            loaded_record['previous'].update(
-                {
-                    'privilegeId': privilege_id,
-                    'compactTransactionId': transaction_id,
-                }
-            )
-            loaded_record.update(
-                {
-                    'compact': TEST_COMPACT,
-                    'jurisdiction': jurisdiction,
-                    'compactTransactionId': transaction_id,
-                    'providerId': licensee_id,
-                }
-            )
+        # Create the previous privilege record
+        previous_privilege = self.test_data_generator.generate_default_privilege(
+            {
+                'privilegeId': privilege_id,
+                'providerId': licensee_id,
+                'compact': TEST_COMPACT,
+                'jurisdiction': jurisdiction,
+                'compactTransactionId': transaction_id,
+            }
+        )
 
-            schema = PrivilegeUpdateRecordSchema()
-            serialized_record = schema.dump(loaded_record)
-            self._provider_table.put_item(Item=serialized_record)
+        # Create the privilege update record
+        # Note: generate_default_privilege_update takes previous_privilege as a separate parameter
+        update_data = self.test_data_generator.generate_default_privilege_update(
+            value_overrides={
+                'compact': TEST_COMPACT,
+                'jurisdiction': jurisdiction,
+                'compactTransactionId': transaction_id,
+                'providerId': licensee_id,
+            },
+            previous_privilege=previous_privilege,
+        )
+        update_record = update_data.serialize_to_database_record()
+        self.test_data_generator.store_record_in_provider_table(update_record)
 
     def _when_testing_non_paginated_event(self, test_compact=TEST_COMPACT):
         return {
@@ -207,7 +204,45 @@ def _when_purchase_client_returns_paginated_transactions(self, mock_purchase_cli
 
         return mock_purchase_client
 
+    def _add_previous_transaction_to_history(
+        self,
+        settlement_time_utc: str = None,
+        transaction_id: str = 'previous-tx-12345',
+        batch_id: str = 'previous-batch-67890',
+    ):
+        """
+        Add a previous transaction to the transaction history table to simulate a previously processed transaction.
+
+        :param settlement_time_utc: Settlement time in UTC (ISO format with Z suffix).
+            If None, defaults to 2 days before scheduled time.
+        :param transaction_id: Transaction ID for the previous transaction
+        :param batch_id: Batch ID for the previous transaction
+        """
+        from cc_common.data_model.transaction_client import TransactionClient
+
+        if settlement_time_utc is None:
+            # Default to 2 days before the scheduled time
+            scheduled_dt = datetime.fromisoformat(MOCK_SCHEDULED_TIME)
+            previous_settlement_dt = scheduled_dt - timedelta(days=2)
+            settlement_time_utc = previous_settlement_dt.replace(tzinfo=None).isoformat() + '.000Z'
+
+        # Format datetime for settlement time local (assuming EST, which is UTC-5)
+        settlement_dt = datetime.fromisoformat(settlement_time_utc)
+        settlement_time_local = (settlement_dt - timedelta(hours=5)).replace(tzinfo=None).strftime('%Y-%m-%dT%H:%M:%S')
+
+        previous_transaction = _generate_mock_transaction(
+            transaction_id=transaction_id,
+            batch_settlement_state='settledSuccessfully',
+        )
+        previous_transaction.batch['batchId'] = batch_id
+        previous_transaction.batch['settlementTimeUTC'] = settlement_time_utc
+        previous_transaction.batch['settlementTimeLocal'] = settlement_time_local
+
+        client = TransactionClient(self.config)
+        client.store_transactions(transactions=[previous_transaction])
+
     @patch('handlers.transaction_history.PurchaseClient')
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-01-01T12:00:00+00:00'))
     def test_process_settled_transactions_returns_complete_status(self, mock_purchase_client_constructor):
         """Test successful processing of settled transactions."""
         from handlers.transaction_history import process_settled_transactions
@@ -217,14 +252,26 @@ def test_process_settled_transactions_returns_complete_status(self, mock_purchas
         self._when_purchase_client_returns_transactions(mock_purchase_client_constructor)
         self._add_mock_privilege_to_database()
         self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
 
         event = self._when_testing_non_paginated_event()
         resp = process_settled_transactions(event, self.mock_context)
 
+        # Calculate expected start/end times
+        scheduled_dt = datetime.fromisoformat(MOCK_SCHEDULED_TIME)
+        previous_settlement_dt = scheduled_dt - timedelta(days=2)
+        expected_start_time = (previous_settlement_dt + timedelta(seconds=1)).strftime('%Y-%m-%dT%H:%M:%SZ')
+        expected_end_time = scheduled_dt.replace(hour=1, minute=0, second=0, microsecond=0).strftime(
+            '%Y-%m-%dT%H:%M:%SZ'
+        )
+
         self.assertEqual(
             {
                 'compact': 'aslp',
                 'scheduledTime': MOCK_SCHEDULED_TIME,
+                'startTime': expected_start_time,
+                'endTime': expected_end_time,
                 'processedBatchIds': [MOCK_BATCH_ID],
                 'status': 'COMPLETE',
             },
@@ -243,6 +290,8 @@ def test_process_settled_transactions_passes_pagination_values_into_purchase_cli
         )
         self._add_mock_privilege_to_database()
         self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
 
         event = self._when_testing_paginated_event()
 
@@ -269,6 +318,8 @@ def test_process_settled_transactions_stores_transactions_in_dynamodb(self, mock
         self._when_purchase_client_returns_transactions(mock_purchase_client_constructor)
         self._add_mock_privilege_to_database()
         self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
 
         event = self._when_testing_non_paginated_event()
 
@@ -284,6 +335,7 @@ def test_process_settled_transactions_stores_transactions_in_dynamodb(self, mock
         # remove dynamic dateOfUpdate timestamp
         del stored_transactions['Items'][0]['dateOfUpdate']
 
+        self.maxDiff = None
         self.assertEqual(
             [
                 {
@@ -335,6 +387,8 @@ def test_process_settled_transactions_does_not_duplicate_identical_transaction_r
         self._when_purchase_client_returns_transactions(mock_purchase_client_constructor)
         self._add_mock_privilege_to_database()
         self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
 
         event = self._when_testing_non_paginated_event()
 
@@ -361,6 +415,7 @@ def test_process_settled_transactions_does_not_duplicate_identical_transaction_r
         self.assertEqual(1, len(stored_transactions['Items']))
 
     @patch('handlers.transaction_history.PurchaseClient')
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-01-01T12:00:00+00:00'))
     def test_process_settled_transactions_returns_in_progress_status_with_pagination_values(
         self, mock_purchase_client_constructor
     ):
@@ -369,14 +424,26 @@ def test_process_settled_transactions_returns_in_progress_status_with_pagination
 
         self._when_purchase_client_returns_paginated_transactions(mock_purchase_client_constructor)
         self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
 
         event = self._when_testing_non_paginated_event()
         resp = process_settled_transactions(event, self.mock_context)
 
+        # Calculate expected start/end times
+        scheduled_dt = datetime.fromisoformat(MOCK_SCHEDULED_TIME)
+        previous_settlement_dt = scheduled_dt - timedelta(days=2)
+        expected_start_time = (previous_settlement_dt + timedelta(seconds=1)).strftime('%Y-%m-%dT%H:%M:%SZ')
+        expected_end_time = scheduled_dt.replace(hour=1, minute=0, second=0, microsecond=0).strftime(
+            '%Y-%m-%dT%H:%M:%SZ'
+        )
+
         self.assertEqual(
             {
                 'compact': TEST_COMPACT,
                 'scheduledTime': MOCK_SCHEDULED_TIME,
+                'startTime': expected_start_time,
+                'endTime': expected_end_time,
                 'status': 'IN_PROGRESS',
                 'lastProcessedTransactionId': MOCK_LAST_PROCESSED_TRANSACTION_ID,
                 'currentBatchId': MOCK_CURRENT_BATCH_ID,
@@ -386,6 +453,7 @@ def test_process_settled_transactions_returns_in_progress_status_with_pagination
         )
 
     @patch('handlers.transaction_history.PurchaseClient')
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-01-01T12:00:00+00:00'))
     def test_process_settled_transactions_returns_batch_failure_status_after_processing_all_transaction(
         self, mock_purchase_client_constructor
     ):
@@ -438,14 +506,26 @@ def test_process_settled_transactions_returns_batch_failure_status_after_process
         ]
 
         self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
         event = self._when_testing_non_paginated_event()
         first_resp = process_settled_transactions(event, self.mock_context)
 
+        # Calculate expected start/end times
+        scheduled_dt = datetime.fromisoformat(MOCK_SCHEDULED_TIME)
+        previous_settlement_dt = scheduled_dt - timedelta(days=2)
+        expected_start_time = (previous_settlement_dt + timedelta(seconds=1)).strftime('%Y-%m-%dT%H:%M:%SZ')
+        expected_end_time = scheduled_dt.replace(hour=1, minute=0, second=0, microsecond=0).strftime(
+            '%Y-%m-%dT%H:%M:%SZ'
+        )
+
         self.assertEqual(
             {
                 'status': 'IN_PROGRESS',
                 'compact': TEST_COMPACT,
                 'scheduledTime': MOCK_SCHEDULED_TIME,
+                'startTime': expected_start_time,
+                'endTime': expected_end_time,
                 'currentBatchId': MOCK_BATCH_ID,
                 'lastProcessedTransactionId': mock_first_iteration_failed_transaction_id,
                 'processedBatchIds': [],
@@ -469,6 +549,8 @@ def test_process_settled_transactions_returns_batch_failure_status_after_process
                 'status': 'BATCH_FAILURE',
                 'compact': TEST_COMPACT,
                 'scheduledTime': MOCK_SCHEDULED_TIME,
+                'startTime': expected_start_time,
+                'endTime': expected_end_time,
                 'processedBatchIds': [MOCK_BATCH_ID],
                 'batchFailureErrorMessage': json.dumps(
                     {
@@ -511,6 +593,8 @@ def test_transaction_with_unknown_privilege_id_in_dynamodb_if_privilege_not_foun
         # privilege id as UNKNOWN
         self._when_purchase_client_returns_transactions(mock_purchase_client_constructor)
         self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
 
         event = self._when_testing_non_paginated_event()
 
@@ -555,6 +639,8 @@ def test_process_settled_transactions_maps_privilege_ids_from_privilege_update_r
         )
 
         self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
         event = self._when_testing_non_paginated_event()
 
         process_settled_transactions(event, self.mock_context)
@@ -628,6 +714,8 @@ def test_process_settled_transactions_maps_privilege_ids_from_privilege_records(
         )
 
         self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
         event = self._when_testing_non_paginated_event()
 
         process_settled_transactions(event, self.mock_context)
@@ -701,9 +789,7 @@ def test_process_settled_transactions_exits_early_when_compact_exists_but_no_jur
         from handlers.transaction_history import process_settled_transactions
 
         # Add only compact configuration data, no jurisdictions
-        with open('../common/tests/resources/dynamo/compact.json') as f:
-            record = json.load(f, parse_float=Decimal)
-            self._compact_configuration_table.put_item(Item=record)
+        self.test_data_generator.put_default_compact_configuration_in_configuration_table({'compactAbbr': TEST_COMPACT})
 
         event = self._when_testing_non_paginated_event()
         resp = process_settled_transactions(event, self.mock_context)
@@ -758,6 +844,8 @@ def test_process_settled_transactions_detects_old_unsettled_transactions(self, m
             transaction_id=MOCK_TRANSACTION_ID,
             privilege_id=MOCK_PRIVILEGE_ID,
         )
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
 
         event = self._when_testing_non_paginated_event()
         resp = process_settled_transactions(event, self.mock_context)
@@ -803,6 +891,8 @@ def test_process_settled_transactions_reconciles_unsettled_transactions(self, mo
             transaction_id=MOCK_TRANSACTION_ID,
             privilege_id=MOCK_PRIVILEGE_ID,
         )
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
 
         event = self._when_testing_non_paginated_event()
         resp = process_settled_transactions(event, self.mock_context)
@@ -817,3 +907,152 @@ def test_process_settled_transactions_reconciles_unsettled_transactions(self, mo
             KeyConditionExpression='pk = :pk', ExpressionAttributeValues={':pk': pk}
         )
         self.assertEqual(0, len(response['Items']))
+
+    @patch('handlers.transaction_history.PurchaseClient')
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-01-01T12:00:00+00:00'))
+    def test_process_settled_transactions_uses_previous_transaction_settlement_time_for_start_time(
+        self, mock_purchase_client_constructor
+    ):
+        """Test that start_time is set to just after the most recent previous transaction's settlement time."""
+        from handlers.transaction_history import process_settled_transactions
+
+        # Set up a previous transaction that is 3 days old
+        scheduled_dt = datetime.fromisoformat(MOCK_SCHEDULED_TIME)
+        previous_settlement_dt = scheduled_dt - timedelta(days=3)
+        previous_settlement_time_utc = previous_settlement_dt.strftime('%Y-%m-%dT%H:%M:%SZ')
+
+        # Add the previous transaction
+        self._add_previous_transaction_to_history(settlement_time_utc=previous_settlement_time_utc)
+
+        self._when_purchase_client_returns_transactions(mock_purchase_client_constructor)
+        self._add_mock_privilege_to_database()
+        self._add_compact_configuration_data()
+
+        event = self._when_testing_non_paginated_event()
+        process_settled_transactions(event, self.mock_context)
+
+        # Verify that get_settled_transactions was called with start_time just after the previous transaction
+        # The start_time should be the previous settlement time plus one second
+        # (since it's more recent than 30 days ago)
+        expected_start_time = (previous_settlement_dt + timedelta(seconds=1)).strftime('%Y-%m-%dT%H:%M:%SZ')
+        mock_purchase_client = mock_purchase_client_constructor.return_value
+        call_kwargs = mock_purchase_client.get_settled_transactions.call_args.kwargs
+        self.assertEqual(expected_start_time, call_kwargs['start_time'])
+
+    @patch('handlers.transaction_history.PurchaseClient')
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-01-01T12:00:00+00:00'))
+    def test_process_settled_transactions_uses_30_day_fallback_when_no_previous_transactions(
+        self, mock_purchase_client_constructor
+    ):
+        """Test that start_time falls back to 30 days ago when no previous transactions exist."""
+        from handlers.transaction_history import process_settled_transactions
+
+        # Don't add any previous transactions - this simulates a compact that just went live
+        self._when_purchase_client_returns_transactions(mock_purchase_client_constructor)
+        self._add_mock_privilege_to_database()
+        self._add_compact_configuration_data()
+
+        event = self._when_testing_non_paginated_event()
+        process_settled_transactions(event, self.mock_context)
+
+        # Verify that get_settled_transactions was called with start_time 30 days before scheduled time
+        # end_time is set to scheduled_time with hour=1, minute=0, second=0, microsecond=0
+        # oldest_allowed_start = end_time - timedelta(days=30) + timedelta(seconds=1)
+        scheduled_dt = datetime.fromisoformat(MOCK_SCHEDULED_TIME)
+        end_time = scheduled_dt.replace(hour=1, minute=0, second=0, microsecond=0)
+        expected_start_time = (end_time - timedelta(days=30)).strftime('%Y-%m-%dT%H:%M:%SZ')
+
+        mock_purchase_client = mock_purchase_client_constructor.return_value
+        call_kwargs = mock_purchase_client.get_settled_transactions.call_args.kwargs
+        self.assertEqual(expected_start_time, call_kwargs['start_time'])
+
+    @patch('handlers.transaction_history.PurchaseClient')
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-01-01T12:00:00+00:00'))
+    def test_process_settled_transactions_uses_provided_start_and_end_times(self, mock_purchase_client_constructor):
+        """
+        Test that when startTime and endTime are provided in the event, they are used instead of calculating new ones.
+        """
+        from handlers.transaction_history import process_settled_transactions
+
+        self._when_purchase_client_returns_transactions(mock_purchase_client_constructor)
+        self._add_mock_privilege_to_database()
+        self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
+
+        # Provide start and end times in the event
+        provided_start_time = '2023-12-15T10:00:00Z'
+        provided_end_time = '2024-01-01T05:00:00Z'
+        event = {
+            'compact': TEST_COMPACT,
+            'scheduledTime': MOCK_SCHEDULED_TIME,
+            'startTime': provided_start_time,
+            'endTime': provided_end_time,
+            'lastProcessedTransactionId': None,
+            'currentBatchId': None,
+            'processedBatchIds': None,
+        }
+
+        resp = process_settled_transactions(event, self.mock_context)
+
+        # Verify that the provided times are used and returned in the response
+        self.assertEqual(provided_start_time, resp['startTime'])
+        self.assertEqual(provided_end_time, resp['endTime'])
+
+        # Verify that get_settled_transactions was called with the provided times
+        mock_purchase_client = mock_purchase_client_constructor.return_value
+        call_kwargs = mock_purchase_client.get_settled_transactions.call_args.kwargs
+        self.assertEqual(provided_start_time, call_kwargs['start_time'])
+        self.assertEqual(provided_end_time, call_kwargs['end_time'])
+
+    @patch('handlers.transaction_history.PurchaseClient')
+    @patch('cc_common.config._Config.current_standard_datetime', datetime.fromisoformat('2024-01-01T12:00:00+00:00'))
+    def test_process_settled_transactions_persists_start_and_end_times_across_pagination_iterations(
+        self, mock_purchase_client_constructor
+    ):
+        """Test that startTime and endTime persist across pagination iterations."""
+        from handlers.transaction_history import process_settled_transactions
+
+        # Set up paginated transactions
+        self._when_purchase_client_returns_paginated_transactions(mock_purchase_client_constructor)
+        self._add_mock_privilege_to_database()
+        self._add_compact_configuration_data()
+        # Add a previous transaction to simulate normal operation
+        self._add_previous_transaction_to_history()
+
+        # First iteration: no start/end times provided, they should be calculated
+        event = self._when_testing_non_paginated_event()
+        resp = process_settled_transactions(event, self.mock_context)
+
+        # Verify start/end times are calculated and returned
+        self.assertIn('startTime', resp)
+        self.assertIn('endTime', resp)
+        first_iteration_start_time = resp['startTime']
+        first_iteration_end_time = resp['endTime']
+
+        # Verify status is IN_PROGRESS
+        self.assertEqual('IN_PROGRESS', resp['status'])
+
+        # Second iteration: use the start/end times from the first iteration
+        event = {
+            'compact': TEST_COMPACT,
+            'scheduledTime': MOCK_SCHEDULED_TIME,
+            'startTime': first_iteration_start_time,
+            'endTime': first_iteration_end_time,
+            'lastProcessedTransactionId': MOCK_LAST_PROCESSED_TRANSACTION_ID,
+            'currentBatchId': MOCK_CURRENT_BATCH_ID,
+            'processedBatchIds': [MOCK_BATCH_ID],
+        }
+
+        resp = process_settled_transactions(event, self.mock_context)
+
+        # Verify that the same start/end times are persisted in the response
+        self.assertEqual(first_iteration_start_time, resp['startTime'])
+        self.assertEqual(first_iteration_end_time, resp['endTime'])
+
+        # Verify that get_settled_transactions was called with the persisted times
+        mock_purchase_client = mock_purchase_client_constructor.return_value
+        # Get the second call (index 1) since we made two calls
+        second_call_kwargs = mock_purchase_client.get_settled_transactions.call_args_list[1].kwargs
+        self.assertEqual(first_iteration_start_time, second_call_kwargs['start_time'])
+        self.assertEqual(first_iteration_end_time, second_call_kwargs['end_time'])
diff --git a/backend/compact-connect/lambdas/python/purchases/tests/unit/test_purchase_client.py b/backend/compact-connect/lambdas/python/purchases/tests/unit/test_purchase_client.py
index 016e848d0..fd20d740c 100644
--- a/backend/compact-connect/lambdas/python/purchases/tests/unit/test_purchase_client.py
+++ b/backend/compact-connect/lambdas/python/purchases/tests/unit/test_purchase_client.py
@@ -1112,11 +1112,12 @@ def test_purchase_client_gets_settled_transactions_successfully(
 
         # Verify transaction data
         transaction = response['transactions'][0]
-        self.assertEqual(transaction['transactionId'], MOCK_TRANSACTION_ID)
-        self.assertEqual(transaction['compact'], 'aslp')
-        self.assertEqual(transaction['licenseeId'], MOCK_LICENSEE_ID)
-        self.assertEqual(transaction['batch']['batchId'], MOCK_BATCH_ID)
-        self.assertEqual(len(transaction['lineItems']), 1)
+
+        self.assertEqual(transaction.transactionId, MOCK_TRANSACTION_ID)
+        self.assertEqual(transaction.compact, 'aslp')
+        self.assertEqual(transaction.licenseeId, MOCK_LICENSEE_ID)
+        self.assertEqual(transaction.batch['batchId'], MOCK_BATCH_ID)
+        self.assertEqual(len(transaction.lineItems), 1)
 
     @patch('purchase_client.getSettledBatchListController')
     @patch('purchase_client.getTransactionListController')
@@ -1197,8 +1198,8 @@ def test_purchase_client_handles_pagination_for_settled_transactions(
         self.assertEqual([MOCK_BATCH_ID], response['processedBatchIds'])
         # Verify transaction data
         self.assertEqual(2, len(response['transactions']))
-        self.assertEqual(response['transactions'][0]['transactionId'], MOCK_TRANSACTION_ID_1_BATCH_1)
-        self.assertEqual(response['transactions'][1]['transactionId'], MOCK_TRANSACTION_ID_1_BATCH_2)
+        self.assertEqual(response['transactions'][0].transactionId, MOCK_TRANSACTION_ID_1_BATCH_1)
+        self.assertEqual(response['transactions'][1].transactionId, MOCK_TRANSACTION_ID_1_BATCH_2)
 
         # now fetch the remaining results
         response = test_purchase_client.get_settled_transactions(
@@ -1218,7 +1219,7 @@ def test_purchase_client_handles_pagination_for_settled_transactions(
         # assert that the second transaction is returned, the first being skipped
         self.assertEqual([MOCK_BATCH_ID, MOCK_BATCH_ID_2], response['processedBatchIds'])
         self.assertEqual(1, len(response['transactions']))
-        self.assertEqual(MOCK_TRANSACTION_ID_2_BATCH_2, response['transactions'][0]['transactionId'])
+        self.assertEqual(MOCK_TRANSACTION_ID_2_BATCH_2, response['transactions'][0].transactionId)
 
     @patch('purchase_client.getSettledBatchListController')
     def test_purchase_client_handles_no_batches_for_settled_transactions(self, mock_batch_controller):
@@ -1290,7 +1291,86 @@ def test_purchase_client_handles_settlement_errors_for_settled_transactions(
 
         # assert that we return the transaction with the settlement error
         self.assertEqual(1, len(response['transactions']))
-        self.assertEqual(MOCK_TRANSACTION_ID, response['transactions'][0]['transactionId'])
-        self.assertEqual(SETTLEMENT_ERROR_STATE, response['transactions'][0]['transactionStatus'])
+        self.assertEqual(MOCK_TRANSACTION_ID, response['transactions'][0].transactionId)
+        self.assertEqual(SETTLEMENT_ERROR_STATE, response['transactions'][0].transactionStatus)
         # assert we return a list of failed transaction ids
         self.assertEqual([MOCK_TRANSACTION_ID], response['settlementErrorTransactionIds'])
+
+    @patch('purchase_client.getSettledBatchListController')
+    @patch('purchase_client.getTransactionListController')
+    @patch('purchase_client.getTransactionDetailsController')
+    def test_purchase_client_skips_declined_transactions(
+        self, mock_details_controller, mock_transaction_controller, mock_batch_controller
+    ):
+        """Test that declined transactions are skipped and not included in the results."""
+        from purchase_client import PurchaseClient
+
+        mock_secrets_manager_client = self._generate_mock_secrets_manager_client()
+
+        # Set up batch with one batch
+        mock_batch_response = json_to_magic_mock(
+            {
+                'messages': {
+                    'resultCode': SUCCESSFUL_RESULT_CODE,
+                },
+                'batchList': {
+                    'batch': [
+                        {
+                            'batchId': MOCK_BATCH_ID,
+                            'settlementTimeUTC': MOCK_SETTLEMENT_TIME_UTC,
+                            'settlementTimeLocal': MOCK_SETTLEMENT_TIME_LOCAL,
+                            'settlementState': SUCCESSFUL_SETTLED_STATE,
+                        }
+                    ]
+                },
+            }
+        )
+        mock_batch_controller.return_value.getresponse.return_value = mock_batch_response
+
+        # Set up transaction list with two transactions: one declined, one successful
+        declined_transaction_id = '999'
+        successful_transaction_id = MOCK_TRANSACTION_ID
+        mock_transaction_list_response = json_to_magic_mock(
+            {
+                'messages': {
+                    'resultCode': SUCCESSFUL_RESULT_CODE,
+                },
+                'transactions': {
+                    'transaction': [
+                        {'transId': declined_transaction_id, 'transactionStatus': 'declined'},
+                        {'transId': successful_transaction_id, 'transactionStatus': SUCCESSFUL_SETTLED_STATE},
+                    ]
+                },
+                'totalNumInResultSet': 2,
+            }
+        )
+        mock_transaction_controller.return_value.getresponse.return_value = mock_transaction_list_response
+
+        # Set up transaction details responses: declined first, then successful
+        declined_details_response = json_to_magic_mock(
+            self._generate_mock_transaction_detail_response(declined_transaction_id, 'declined')
+        )
+        successful_details_response = json_to_magic_mock(
+            self._generate_mock_transaction_detail_response(successful_transaction_id, SUCCESSFUL_SETTLED_STATE)
+        )
+        mock_details_controller.return_value.getresponse.side_effect = [
+            declined_details_response,
+            successful_details_response,
+        ]
+
+        test_purchase_client = PurchaseClient(secrets_manager_client=mock_secrets_manager_client)
+        response = test_purchase_client.get_settled_transactions(
+            compact='aslp',
+            start_time='2024-01-01T00:00:00Z',
+            end_time='2024-01-02T00:00:00Z',
+            transaction_limit=500,
+        )
+
+        # Verify only the successful transaction is returned (declined one is skipped)
+        self.assertEqual(1, len(response['transactions']))
+        self.assertEqual(successful_transaction_id, response['transactions'][0].transactionId)
+        self.assertEqual(SUCCESSFUL_SETTLED_STATE, response['transactions'][0].transactionStatus)
+        # Verify declined transaction is not in the results
+        transaction_ids = [tx.transactionId for tx in response['transactions']]
+        self.assertNotIn(declined_transaction_id, transaction_ids)
+        self.assertIn(successful_transaction_id, transaction_ids)
diff --git a/backend/compact-connect/stacks/transaction_monitoring_stack/transaction_history_processing_workflow.py b/backend/compact-connect/stacks/transaction_monitoring_stack/transaction_history_processing_workflow.py
index a2a909c6b..9ebef371b 100644
--- a/backend/compact-connect/stacks/transaction_monitoring_stack/transaction_history_processing_workflow.py
+++ b/backend/compact-connect/stacks/transaction_monitoring_stack/transaction_history_processing_workflow.py
@@ -216,12 +216,12 @@ def __init__(
         )
 
         # By default, the authorize.net accounts batch settlements at 4:00pm Pacific Time.
-        # This daily collector runs an hour later (5pm PST, which is 1am UTC) to collect
-        # all settled transaction for the last 24 hours.
+        # In practice, we've seen settlements running up to an hour late, so this daily collector runs two hours later
+        # (6pm PST, which is 2am UTC) to collect all settled transactions since the previous run.
         Rule(
             self,
             f'{compact}-DailyTransactionProcessingRule',
-            schedule=Schedule.cron(week_day='*', hour='1', minute='0', month='*', year='*'),
+            schedule=Schedule.cron(week_day='*', hour='2', minute='0', month='*', year='*'),
             targets=[SfnStateMachine(state_machine)],
         )