diff --git a/django-backend/soroscan/ingest/management/commands/import_contracts.py b/django-backend/soroscan/ingest/management/commands/import_contracts.py
new file mode 100644
index 00000000..b563398f
--- /dev/null
+++ b/django-backend/soroscan/ingest/management/commands/import_contracts.py
@@ -0,0 +1,93 @@
+"""
+Management command: import_contracts
+
+Reads a JSON file with address/name mappings and creates TrackedContract records.
+Skips contracts that already exist (by contract_id).
+
+Usage:
+    python manage.py import_contracts --file contracts.json
+
+JSON format:
+    {"contracts": [{"address": "C...", "name": "My Contract"}, ...]}
+    or a top-level list: [{"address": "C...", "name": "My Contract"}, ...]
+"""
+import json
+
+from django.contrib.auth import get_user_model
+from django.core.management.base import BaseCommand, CommandError
+
+from soroscan.ingest.models import TrackedContract
+
+User = get_user_model()
+
+
+class Command(BaseCommand):
+    help = "Bulk import contracts from a JSON file (skips duplicates)."
+
+    def add_arguments(self, parser):
+        parser.add_argument("--file", required=True, help="Path to JSON file")
+        parser.add_argument(
+            "--owner",
+            default=None,
+            help="Username to assign as owner (defaults to first superuser)",
+        )
+
+    def handle(self, *args, **options):
+        file_path = options["file"]
+        owner_username = options["owner"]
+
+        try:
+            with open(file_path, encoding="utf-8") as f:
+                data = json.load(f)
+        except FileNotFoundError:
+            raise CommandError(f"File not found: {file_path}")
+        except json.JSONDecodeError as exc:
+            raise CommandError(f"Invalid JSON: {exc}")
+
+        contracts = data if isinstance(data, list) else data.get("contracts", [])
+        if not isinstance(contracts, list):
+            raise CommandError("JSON must be a list or an object with a 'contracts' list.")
+
+        # Resolve owner
+        if owner_username:
+            try:
+                owner = User.objects.get(username=owner_username)
+            except User.DoesNotExist:
+                raise CommandError(f"User '{owner_username}' not found.")
+        else:
+            owner = User.objects.filter(is_superuser=True).first()
+            if owner is None:
+                owner = User.objects.first()
+            if owner is None:
+                raise CommandError(
+                    "No users exist. Create a user first or pass --owner."
+                )
+
+        created_count = 0
+        skipped_count = 0
+
+        for entry in contracts:
+            address = entry.get("address") or entry.get("contract_id", "")
+            name = entry.get("name", "")
+
+            if not address:
+                self.stderr.write(self.style.WARNING(f"Skipping entry with no address: {entry}"))
+                skipped_count += 1
+                continue
+
+            _, created = TrackedContract.objects.get_or_create(
+                contract_id=address,
+                defaults={"name": name or address, "owner": owner},
+            )
+            if created:
+                created_count += 1
+                self.stdout.write(f"  Created: {address} ({name})")
+            else:
+                skipped_count += 1
+                self.stdout.write(f"  Skipped (exists): {address}")
+
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"Done. Created: {created_count}, Skipped: {skipped_count}"
+            )
+        )
diff --git a/django-backend/soroscan/ingest/migrations/0041_blacklistedcontract.py b/django-backend/soroscan/ingest/migrations/0041_blacklistedcontract.py
new file mode 100644
index 00000000..b1814dbc
--- /dev/null
+++ b/django-backend/soroscan/ingest/migrations/0041_blacklistedcontract.py
@@ -0,0 +1,25 @@
+# Generated migration for BlacklistedContract model
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("ingest", "0040_alter_trackedcontract_contract_id"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="BlacklistedContract",
+            fields=[
+                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("contract_id", models.CharField(db_index=True, help_text="Stellar contract address to block from indexing", max_length=56, unique=True)),
+                ("reason", models.TextField(blank=True, help_text="Optional reason for blacklisting")),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+            ],
+            options={
+                "ordering": ["-created_at"],
+            },
+        ),
+    ]
diff --git a/django-backend/soroscan/ingest/models.py b/django-backend/soroscan/ingest/models.py
index 85352e05..641ba159 100644
--- a/django-backend/soroscan/ingest/models.py
+++ b/django-backend/soroscan/ingest/models.py
@@ -240,6 +240,27 @@ def __str__(self):
         return f"{self.user} @ {self.team} ({self.role})"
 
 
+class BlacklistedContract(models.Model):
+    """
+    Contracts whose events should never be indexed (spam prevention).
+    """
+
+    contract_id = models.CharField(
+        max_length=56,
+        unique=True,
+        db_index=True,
+        help_text="Stellar contract address to block from indexing",
+    )
+    reason = models.TextField(blank=True, help_text="Optional reason for blacklisting")
+    created_at = models.DateTimeField(auto_now_add=True)
+
+    class Meta:
+        ordering = ["-created_at"]
+
+    def __str__(self):
+        return f"Blacklisted: {self.contract_id}"
+
+
 class TrackedContract(models.Model):
     """
     Contracts registered for event indexing.
diff --git a/django-backend/soroscan/ingest/tasks.py b/django-backend/soroscan/ingest/tasks.py
index 07e02d6e..29ce4af3 100644
--- a/django-backend/soroscan/ingest/tasks.py
+++ b/django-backend/soroscan/ingest/tasks.py
@@ -56,6 +56,7 @@
     OrganizationBudget,
     OrganizationCostSnapshot,
     WebhookDeadLetter,
+    BlacklistedContract,
 )
 from .rate_limit import check_ingest_rate
 from .stellar_client import SorobanClient
@@ -2017,6 +2018,20 @@ def ingest_latest_events() -> int:
                 ).inc()
                 continue
 
+            # Skip blacklisted contracts
+            if BlacklistedContract.objects.filter(contract_id=event.contract_id).exists():
+                logger.info(
+                    "Skipping blacklisted contract %s",
+                    event.contract_id,
+                    extra={"contract_id": event.contract_id},
+                )
+                m.events_skipped_total.labels(
+                    contract_id=_short_contract_id(event.contract_id),
+                    network=network,
+                    reason="blacklisted",
+                ).inc()
+                continue
+
             # Check rate limit before processing
             if not check_ingest_rate(contract):
                 m.events_rate_limited_total.labels(
diff --git a/django-backend/soroscan/ingest/tests/test_four_features.py b/django-backend/soroscan/ingest/tests/test_four_features.py
new file mode 100644
index 00000000..edb3df54
--- /dev/null
+++ b/django-backend/soroscan/ingest/tests/test_four_features.py
@@ -0,0 +1,290 @@
+"""
+Tests for the four new features:
+  1. import_contracts management command
+  2. GET /api/webhooks/schema/ endpoint
+  3. BlacklistedContract model and ingestion skip
+  4. ?type= filter on ContractEventViewSet
+"""
+import json
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+from django.contrib.auth import get_user_model
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from rest_framework.test import APIClient
+
+from soroscan.ingest.models import BlacklistedContract, ContractEvent, TrackedContract
+from soroscan.ingest.tests.factories import (
+    ContractEventFactory,
+    TrackedContractFactory,
+    UserFactory,
+)
+
+User = get_user_model()
+
+
+# ---------------------------------------------------------------------------
+# Feature 1: import_contracts management command
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.django_db
+class TestImportContractsCommand:
+    def test_imports_contracts_from_json_list(self):
+        user = UserFactory()
+        contracts_data = [
+            {"address": "C" + "A" * 55, "name": "Contract A"},
+            {"address": "C" + "B" * 55, "name": "Contract B"},
+        ]
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+            json.dump(contracts_data, f)
+            f.flush()
+
+            call_command("import_contracts", file=f.name, owner=user.username)
+
+        assert TrackedContract.objects.filter(contract_id="C" + "A" * 55).exists()
+        assert TrackedContract.objects.filter(contract_id="C" + "B" * 55).exists()
+        assert TrackedContract.objects.count() == 2
+
+    def test_imports_contracts_from_json_object(self):
+        user = UserFactory()
+        contracts_data = {
+            "contracts": [
+                {"address": "C" + "C" * 55, "name": "Contract C"},
+                {"address": "C" + "D" * 55, "name": "Contract D"},
+            ]
+        }
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+            json.dump(contracts_data, f)
+            f.flush()
+
+            call_command("import_contracts", file=f.name, owner=user.username)
+
+        assert TrackedContract.objects.filter(contract_id="C" + "C" * 55).exists()
+        assert TrackedContract.objects.filter(contract_id="C" + "D" * 55).exists()
+
+    def test_skips_duplicate_contracts(self):
+        user = UserFactory()
+        existing = TrackedContractFactory(contract_id="C" + "E" * 55, owner=user)
+
+        contracts_data = [
+            {"address": "C" + "E" * 55, "name": "Duplicate"},
+            {"address": "C" + "F" * 55, "name": "New Contract"},
+        ]
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+            json.dump(contracts_data, f)
+            f.flush()
+
+            call_command("import_contracts", file=f.name, owner=user.username)
+
+        assert TrackedContract.objects.count() == 2
+        assert TrackedContract.objects.filter(contract_id="C" + "F" * 55).exists()
+
+    def test_handles_missing_file(self):
+        with pytest.raises(CommandError, match="File not found"):
+            call_command("import_contracts", file="/nonexistent.json")
+
+    def test_handles_invalid_json(self):
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+            f.write("not valid json{")
+            f.flush()
+
+            with pytest.raises(CommandError, match="Invalid JSON"):
+                call_command("import_contracts", file=f.name)
+
+    def test_uses_first_superuser_when_no_owner_specified(self):
+        superuser = UserFactory(is_superuser=True)
+        contracts_data = [{"address": "C" + "G" * 55, "name": "Contract G"}]
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+            json.dump(contracts_data, f)
+            f.flush()
+
+            call_command("import_contracts", file=f.name)
+
+        contract = TrackedContract.objects.get(contract_id="C" + "G" * 55)
+        assert contract.owner == superuser
+
+
+# ---------------------------------------------------------------------------
+# Feature 2: GET /api/webhooks/schema/ endpoint
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.django_db
+class TestWebhookSchemaEndpoint:
+    def test_returns_json_schema(self):
+        client = APIClient()
+        response = client.get("/api/ingest/webhooks/schema/")
+
+        assert response.status_code == 200
+        schema = response.json()
+        assert schema["$schema"] == "http://json-schema.org/draft-07/schema#"
+        assert schema["title"] == "SoroScan Webhook Event Payload"
+        assert "properties" in schema
+        assert "contract_id" in schema["properties"]
+        assert "event_type" in schema["properties"]
+        assert "ledger" in schema["properties"]
+        assert "payload" in schema["properties"]
+
+    def test_schema_has_required_fields(self):
+        client = APIClient()
+        response = client.get("/api/ingest/webhooks/schema/")
+
+        schema = response.json()
+        assert "required" in schema
+        assert "contract_id" in schema["required"]
+        assert "event_type" in schema["required"]
+        assert "ledger" in schema["required"]
+        assert "timestamp" in schema["required"]
+        assert "payload" in schema["required"]
+
+    def test_schema_describes_contract_id_pattern(self):
+        client = APIClient()
+        response = client.get("/api/ingest/webhooks/schema/")
+
+        schema = response.json()
+        contract_id_prop = schema["properties"]["contract_id"]
+        assert contract_id_prop["type"] == "string"
+        assert "pattern" in contract_id_prop
+        assert contract_id_prop["pattern"] == "^C[A-Z2-7]{55}$"
+
+
+# ---------------------------------------------------------------------------
+# Feature 3: BlacklistedContract model and ingestion skip
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.django_db
+class TestBlacklistedContract:
+    def test_model_creation(self):
+        blacklisted = BlacklistedContract.objects.create(
+            contract_id="C" + "X" * 55,
+            reason="Spam contract",
+        )
+        assert blacklisted.contract_id == "C" + "X" * 55
+        assert blacklisted.reason == "Spam contract"
+        assert blacklisted.created_at is not None
+
+    def test_unique_constraint(self):
+        BlacklistedContract.objects.create(contract_id="C" + "Y" * 55)
+        with pytest.raises(Exception):  # IntegrityError
+            BlacklistedContract.objects.create(contract_id="C" + "Y" * 55)
+
+    @patch("stellar_sdk.SorobanServer")
+    def test_ingestion_skips_blacklisted_contracts(self, mock_server_class):
+        """Test that blacklisted contracts are skipped during ingestion."""
+        from soroscan.ingest.tasks import ingest_latest_events
+
+        # Create a tracked contract and blacklist it
+        contract = TrackedContractFactory(contract_id="C" + "Z" * 55)
+        BlacklistedContract.objects.create(contract_id=contract.contract_id)
+
+        # Mock the SorobanServer to return an event for the blacklisted contract
+        mock_event = MagicMock()
+        mock_event.contract_id = contract.contract_id
+        mock_event.type = "transfer"
+        mock_event.value = {"amount": 100}
+        mock_event.ledger = 1000
+        mock_event.tx_hash = "abc123"
+
+        mock_response = MagicMock()
+        mock_response.events = [mock_event]
+        mock_response.cursor = "1000"
+
+        mock_server = MagicMock()
+        mock_server.get_events.return_value = mock_response
+        mock_server_class.return_value = mock_server
+
+        ingest_latest_events()
+
+        # Verify no events were created for the blacklisted contract
+        assert ContractEvent.objects.filter(contract=contract).count() == 0
+
+
+# ---------------------------------------------------------------------------
+# Feature 4: ?type= filter on ContractEventViewSet
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.django_db
+class TestEventTypeFilter:
+    def test_filters_by_single_type(self):
+        user = UserFactory()
+        contract = TrackedContractFactory(owner=user)
+        ContractEventFactory(contract=contract, event_type="transfer")
+        ContractEventFactory(contract=contract, event_type="swap")
+        ContractEventFactory(contract=contract, event_type="burn")
+
+        client = APIClient()
+        client.force_authenticate(user=user)
+        response = client.get("/api/ingest/events/?type=transfer")
+
+        assert response.status_code == 200
+        results = response.json()["results"]
+        assert len(results) == 1
+        assert results[0]["event_type"] == "transfer"
+
+    def test_filters_by_multiple_types(self):
+        user = UserFactory()
+        contract = TrackedContractFactory(owner=user)
+        ContractEventFactory(contract=contract, event_type="transfer")
+        ContractEventFactory(contract=contract, event_type="swap")
+        ContractEventFactory(contract=contract, event_type="burn")
+        ContractEventFactory(contract=contract, event_type="mint")
+
+        client = APIClient()
+        client.force_authenticate(user=user)
+        response = client.get("/api/ingest/events/?type=transfer,burn")
+
+        assert response.status_code == 200
+        results = response.json()["results"]
+        assert len(results) == 2
+        event_types = {r["event_type"] for r in results}
+        assert event_types == {"transfer", "burn"}
+
+    def test_returns_empty_list_when_no_matches(self):
+        user = UserFactory()
+        contract = TrackedContractFactory(owner=user)
+        ContractEventFactory(contract=contract, event_type="transfer")
+
+        client = APIClient()
+        client.force_authenticate(user=user)
+        response = client.get("/api/ingest/events/?type=nonexistent")
+
+        assert response.status_code == 200
+        results = response.json()["results"]
+        assert len(results) == 0
+
+    def test_ignores_empty_type_parameter(self):
+        user = UserFactory()
+        contract = TrackedContractFactory(owner=user)
+        ContractEventFactory(contract=contract, event_type="transfer")
+        ContractEventFactory(contract=contract, event_type="swap")
+
+        client = APIClient()
+        client.force_authenticate(user=user)
+        response = client.get("/api/ingest/events/?type=")
+
+        assert response.status_code == 200
+        results = response.json()["results"]
+        assert len(results) == 2
+
+    def test_handles_whitespace_in_type_list(self):
+        user = UserFactory()
+        contract = TrackedContractFactory(owner=user)
+        ContractEventFactory(contract=contract, event_type="transfer")
+        ContractEventFactory(contract=contract, event_type="swap")
+
+        client = APIClient()
+        client.force_authenticate(user=user)
+        response = client.get("/api/ingest/events/?type=transfer, swap")
+
+        assert response.status_code == 200
+        results = response.json()["results"]
+        assert len(results) == 2
diff --git a/django-backend/soroscan/ingest/tests/test_migration_graph.py b/django-backend/soroscan/ingest/tests/test_migration_graph.py
index b038f488..4ce52152 100644
--- a/django-backend/soroscan/ingest/tests/test_migration_graph.py
+++ b/django-backend/soroscan/ingest/tests/test_migration_graph.py
@@ -21,7 +21,7 @@ def test_single_leaf_node():
     """
     Assert the ingest migration graph has exactly one leaf node.
 
-    The current leaf is '0040_alter_trackedcontract_contract_id'
+    The current leaf is '0041_blacklistedcontract'
     """
     loader = MigrationLoader(None, ignore_no_migrations=True)
 
@@ -31,8 +31,8 @@ def test_single_leaf_node():
     assert len(leaf_nodes) == 1, (
         f"Expected 1 leaf node for 'ingest', found {len(leaf_nodes)}: {leaf_nodes}"
     )
-    assert leaf_nodes[0][1] == "0040_alter_trackedcontract_contract_id", (
-        "Expected leaf node '0040_alter_trackedcontract_contract_id', "
+    assert leaf_nodes[0][1] == "0041_blacklistedcontract", (
+        "Expected leaf node '0041_blacklistedcontract', "
         f"got '{leaf_nodes[0][1]}'"
     )
 
diff --git a/django-backend/soroscan/ingest/views.py b/django-backend/soroscan/ingest/views.py
index 02b1bcb9..a1a25003 100644
--- a/django-backend/soroscan/ingest/views.py
+++ b/django-backend/soroscan/ingest/views.py
@@ -310,7 +310,13 @@ class ContractEventViewSet(viewsets.ReadOnlyModelViewSet):
     ordering = ["-timestamp"]
 
     def get_queryset(self):
-        return ContractEvent.objects.select_related("contract").all()
+        qs = ContractEvent.objects.select_related("contract").all()
+        type_param = self.request.query_params.get("type", "").strip()
+        if type_param:
+            types = [t.strip() for t in type_param.split(",") if t.strip()]
+            if types:
+                qs = qs.filter(event_type__in=types)
+        return qs
 
     @extend_schema(
         parameters=[
@@ -648,6 +654,61 @@ def dry_run(self, request, pk=None):
         matched = evaluate_condition(webhook.filter_condition, sample_event)
         return Response({"matched": bool(matched)})
 
+    @extend_schema(
+        responses={
+            200: inline_serializer(
+                name="WebhookEventPayloadSchema",
+                fields={"schema": serializers.JSONField()},
+            )
+        },
+    )
+    @action(detail=False, methods=["get"], url_path="schema", url_name="event-schema", permission_classes=[AllowAny])
+    def event_schema(self, request):
+        """Return a JSON Schema describing the webhook event payload structure."""
+        payload_schema = {
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "title": "SoroScan Webhook Event Payload",
+            "type": "object",
+            "required": ["contract_id", "event_type", "ledger", "timestamp", "payload"],
+            "properties": {
+                "contract_id": {
+                    "type": "string",
+                    "description": "Stellar contract address (C...)",
+                    "pattern": "^C[A-Z2-7]{55}$",
+                },
+                "event_type": {
+                    "type": "string",
+                    "description": "Type of the emitted event (e.g. 'transfer', 'swap')",
+                },
+                "ledger": {
+                    "type": "integer",
+                    "description": "Ledger sequence number when the event was emitted",
+                },
+                "event_index": {
+                    "type": "integer",
+                    "description": "Position of the event within the ledger",
+                },
+                "tx_hash": {
+                    "type": "string",
+                    "description": "Transaction hash that produced the event",
+                },
+                "timestamp": {
+                    "type": "string",
+                    "format": "date-time",
+                    "description": "ISO 8601 timestamp of the event",
+                },
+                "payload": {
+                    "type": "object",
+                    "description": "Decoded event payload (contract-specific fields)",
+                },
+                "signature": {
+                    "type": "string",
+                    "description": "HMAC-SHA256 signature header value (sha256=<hex>)",
+                },
+            },
+        }
+        return Response(payload_schema)
+
 
 class TeamViewSet(viewsets.ModelViewSet):
     """