diff --git a/CHANGELOG.md b/CHANGELOG.md
index d4bde94f..9863cc47 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -55,6 +55,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Blazegraph literal serialization** (PR #448, @KaifAhmad1) — `_format_object_for_sparql()` selects IRI/typed-literal/language-tagged-literal/plain-literal token; `_resolve_datatype_iri()` with prefix expansion; RFC 5646 language-tag validation; `_escape_literal()` for string escaping.
 - **DeepSeek provider via OpenAI SDK** (PR #482, @liling) — `_init_client` rewritten using `openai.OpenAI(base_url=self.base_url)` instead of defunct `deepseek` package; `verbose_mode` assignment fix; `pyproject.toml` updated to `openai>=1.0.0`.
 
+### Added
+
+- **`DuplicateDetector` result limiting and ranking** (issue #534, by @KaifAhmad1):
+  - `max_results` — hard global cap on returned candidates; applied after sorting. `None` means no limit.
+  - `top_k_per_entity` — keep at most *k* candidates per entity (by the sort field) so no single entity floods the output. `None` means no per-entity limit.
+  - `min_similarity` — extra similarity floor on top of `similarity_threshold`; candidates below it are dropped before ranking. `None` means no extra floor.
+  - `sort_by` — ranking field before limits are applied; accepts `"confidence"` (default) or `"similarity_score"`. Invalid values raise `ValueError` at construction time.
+  - All four options are applied by the new `_apply_result_limits` helper and are respected by both `detect_duplicates()` and `incremental_detect()`.
+  - 15 new tests in `TestResultLimiting` covering each option in isolation and in combination.
+
 ### Fixed
 
 - **Fix: `ConflictDetector.detect_conflicts()` raises `AttributeError` when called with `method=` or `property_name=` kwargs** (issue #533, PR conflicts, by @KaifAhmad1):
diff --git a/semantica/deduplication/duplicate_detector.py b/semantica/deduplication/duplicate_detector.py
index fc89dddd..52710b8e 100644
--- a/semantica/deduplication/duplicate_detector.py
+++ b/semantica/deduplication/duplicate_detector.py
@@ -100,6 +100,10 @@ class DuplicateDetector:
         confidence_threshold: float = 0.6,
         use_clustering: bool = True,
         config: Optional[Dict[str, Any]] = None,
+        max_results: Optional[int] = None,
+        top_k_per_entity: Optional[int] = None,
+        min_similarity: Optional[float] = None,
+        sort_by: str = "confidence",
         **kwargs,
     ):
         """
@@ -115,6 +119,16 @@ class DuplicateDetector:
                                  (0.0 to 1.0, default: 0.6)
             use_clustering: Whether to use clustering for group formation (default: True)
             config: Configuration dictionary (merged with kwargs)
+            max_results: Hard cap on total candidates returned across all entities.
+                         Applied after sorting. ``None`` means no limit.
+            top_k_per_entity: Keep at most this many candidates per entity (by the
+                              sort field). ``None`` means no per-entity limit.
+            min_similarity: Additional similarity floor applied on top of
+                            ``similarity_threshold``. Candidates whose
+                            ``similarity_score`` is below this value are dropped
+                            before ranking. ``None`` means no extra floor.
+            sort_by: Field used for ranking before limits are applied.
+                     ``"confidence"`` (default) or ``"similarity_score"``.
             **kwargs: Additional configuration options:
                 - similarity: Configuration for SimilarityCalculator
         """
@@ -133,6 +147,16 @@ class DuplicateDetector:
         self.confidence_threshold = confidence_threshold
         self.use_clustering = use_clustering
 
+        # Result limiting / ranking options
+        self.max_results = max_results
+        self.top_k_per_entity = top_k_per_entity
+        self.min_similarity = min_similarity
+        if sort_by not in ("confidence", "similarity_score"):
+            raise ValueError(
+                f"sort_by must be 'confidence' or 'similarity_score', got {sort_by!r}"
+            )
+        self.sort_by = sort_by
+
         # Initialize progress tracker and ensure it's enabled
         self.progress_tracker = get_progress_tracker()
         if not self.progress_tracker.enabled:
@@ -140,7 +164,9 @@ class DuplicateDetector:
 
         self.logger.debug(
             f"Duplicate detector initialized: similarity_threshold={similarity_threshold}, "
-            f"confidence_threshold={confidence_threshold}"
+            f"confidence_threshold={confidence_threshold}, max_results={max_results}, "
+            f"top_k_per_entity={top_k_per_entity}, min_similarity={min_similarity}, "
+            f"sort_by={sort_by!r}"
         )
 
     def detect_duplicates(
@@ -255,8 +281,8 @@ class DuplicateDetector:
                         message=f"Creating duplicate candidates... {i + 1}/{total_similarities} (remaining: {remaining})",
                     )
 
-            # Sort by confidence (highest first)
-            candidates.sort(key=lambda c: c.confidence, reverse=True)
+            # Sort, filter, and cap results
+            candidates = self._apply_result_limits(candidates)
 
             self.logger.info(
                 f"Detected {len(candidates)} duplicate candidate(s) "
@@ -600,8 +626,8 @@ class DuplicateDetector:
                             message=f"Comparing entities... {processed}/{total_comparisons} (remaining: {remaining})",
                         )
 
-            # Sort by confidence (highest first)
-            candidates.sort(key=lambda c: c.confidence, reverse=True)
+            # Sort, filter, and cap results
+            candidates = self._apply_result_limits(candidates)
 
             self.logger.info(
                 f"Incremental detection found {len(candidates)} duplicate candidate(s)"
@@ -620,6 +646,49 @@ class DuplicateDetector:
             )
             raise
 
+    def _apply_result_limits(
+        self, candidates: List[DuplicateCandidate]
+    ) -> List[DuplicateCandidate]:
+        """
+        Apply min_similarity filter, sort, top_k_per_entity, and max_results cap.
+
+        Order of operations:
+        1. Drop candidates below ``min_similarity`` (if set).
+        2. Sort by ``sort_by`` field descending.
+        3. Apply ``top_k_per_entity``: for each entity id keep only the top-k
+           candidates in which it appears.
+        4. Apply ``max_results`` global cap.
+        """
+        # 1. min_similarity filter
+        if self.min_similarity is not None:
+            candidates = [
+                c for c in candidates if c.similarity_score >= self.min_similarity
+            ]
+
+        # 2. Sort descending by the chosen field
+        candidates.sort(key=lambda c: getattr(c, self.sort_by), reverse=True)
+
+        # 3. top_k_per_entity
+        if self.top_k_per_entity is not None:
+            entity_counts: Dict[str, int] = {}
+            kept: List[DuplicateCandidate] = []
+            for c in candidates:
+                id1 = self._get_entity_value(c.entity1, "id") or id(c.entity1)
+                id2 = self._get_entity_value(c.entity2, "id") or id(c.entity2)
+                count1 = entity_counts.get(str(id1), 0)
+                count2 = entity_counts.get(str(id2), 0)
+                if count1 < self.top_k_per_entity and count2 < self.top_k_per_entity:
+                    kept.append(c)
+                    entity_counts[str(id1)] = count1 + 1
+                    entity_counts[str(id2)] = count2 + 1
+            candidates = kept
+
+        # 4. max_results global cap
+        if self.max_results is not None:
+            candidates = candidates[: self.max_results]
+
+        return candidates
+
     def _get_entity_value(self, entity: Any, key: str, default: Any = None) -> Any:
         """Get value from entity dictionary or object safely."""
         if hasattr(entity, "__dict__"):
diff --git a/tests/deduplication/test_deduplication.py b/tests/deduplication/test_deduplication.py
index c6b224e6..b538adac 100644
--- a/tests/deduplication/test_deduplication.py
+++ b/tests/deduplication/test_deduplication.py
@@ -256,5 +256,258 @@ class TestProgressTrackerEncoding(unittest.TestCase):
             sys.stdout = orig
 
 
+class TestResultLimiting(unittest.TestCase):
+    """Tests for issue #534 — max_results, top_k_per_entity, min_similarity, sort_by."""
+
+    def setUp(self):
+        # Six entities: three near-duplicate Apple variants + two Microsoft variants + one Google.
+        # Lower thresholds so all intra-brand pairs clear the bar.
+        self.entities = [
+            {"id": "a1", "name": "Apple Inc.", "type": "Company",
+             "properties": {"industry": "Technology"}},
+            {"id": "a2", "name": "Apple", "type": "Company",
+             "properties": {"industry": "Tech"}},
+            {"id": "a3", "name": "Apple Corp", "type": "Company",
+             "properties": {"industry": "Technology"}},
+            {"id": "b1", "name": "Microsoft Corporation", "type": "Company",
+             "properties": {"industry": "Software"}},
+            {"id": "b2", "name": "Microsoft Corp", "type": "Company",
+             "properties": {"industry": "Software"}},
+            {"id": "c1", "name": "Google LLC", "type": "Company",
+             "properties": {"industry": "Internet"}},
+        ]
+        self.threshold = 0.3
+
+    def _base_detector(self, **kwargs):
+        return DuplicateDetector(
+            similarity_threshold=self.threshold,
+            confidence_threshold=self.threshold,
+            **kwargs,
+        )
+
+    # ------------------------------------------------------------------
+    # max_results
+    # ------------------------------------------------------------------
+
+    def test_max_results_caps_output(self):
+        detector = self._base_detector(max_results=1)
+        results = detector.detect_duplicates(self.entities)
+        self.assertLessEqual(len(results), 1)
+
+    def test_max_results_two(self):
+        detector = self._base_detector(max_results=2)
+        results = detector.detect_duplicates(self.entities)
+        self.assertLessEqual(len(results), 2)
+
+    def test_max_results_none_no_cap(self):
+        uncapped = self._base_detector()
+        large_cap = self._base_detector(max_results=999)
+        self.assertEqual(
+            len(uncapped.detect_duplicates(self.entities)),
+            len(large_cap.detect_duplicates(self.entities)),
+        )
+
+    def test_max_results_zero_returns_empty(self):
+        detector = self._base_detector(max_results=0)
+        self.assertEqual(detector.detect_duplicates(self.entities), [])
+
+    def test_max_results_returns_highest_confidence_first(self):
+        """When capped, the kept candidates must be the highest-confidence ones."""
+        n = 2
+        all_results = self._base_detector().detect_duplicates(self.entities)
+        capped = self._base_detector(max_results=n).detect_duplicates(self.entities)
+        if len(all_results) >= n:
+            expected_ids = {
+                (c.entity1["id"], c.entity2["id"]) for c in all_results[:n]
+            }
+            actual_ids = {
+                (c.entity1["id"], c.entity2["id"]) for c in capped
+            }
+            self.assertEqual(expected_ids, actual_ids)
+
+    def test_max_results_empty_input(self):
+        detector = self._base_detector(max_results=5)
+        self.assertEqual(detector.detect_duplicates([]), [])
+
+    # ------------------------------------------------------------------
+    # top_k_per_entity
+    # ------------------------------------------------------------------
+
+    def test_top_k_per_entity_k1(self):
+        k = 1
+        results = self._base_detector(top_k_per_entity=k).detect_duplicates(self.entities)
+        counts: Dict[str, int] = {}
+        for c in results:
+            for eid in (c.entity1["id"], c.entity2["id"]):
+                counts[eid] = counts.get(eid, 0) + 1
+        for eid, count in counts.items():
+            self.assertLessEqual(count, k, f"Entity {eid!r} appears {count} times, expected <= {k}")
+
+    def test_top_k_per_entity_k2(self):
+        k = 2
+        results = self._base_detector(top_k_per_entity=k).detect_duplicates(self.entities)
+        counts: Dict[str, int] = {}
+        for c in results:
+            for eid in (c.entity1["id"], c.entity2["id"]):
+                counts[eid] = counts.get(eid, 0) + 1
+        for eid, count in counts.items():
+            self.assertLessEqual(count, k)
+
+    def test_top_k_per_entity_large_k_same_as_none(self):
+        uncapped = self._base_detector().detect_duplicates(self.entities)
+        large_k = self._base_detector(top_k_per_entity=999).detect_duplicates(self.entities)
+        self.assertEqual(len(uncapped), len(large_k))
+
+    def test_top_k_per_entity_empty_input(self):
+        detector = self._base_detector(top_k_per_entity=2)
+        self.assertEqual(detector.detect_duplicates([]), [])
+
+    # ------------------------------------------------------------------
+    # min_similarity
+    # ------------------------------------------------------------------
+
+    def test_min_similarity_all_results_above_floor(self):
+        floor = 0.6
+        results = self._base_detector(min_similarity=floor).detect_duplicates(self.entities)
+        for c in results:
+            self.assertGreaterEqual(
+                c.similarity_score, floor,
+                f"Candidate score {c.similarity_score} is below min_similarity={floor}",
+            )
+
+    def test_min_similarity_very_high_returns_only_exact(self):
+        results = self._base_detector(min_similarity=1.0).detect_duplicates(self.entities)
+        for c in results:
+            self.assertEqual(c.similarity_score, 1.0)
+
+    def test_min_similarity_zero_does_not_over_filter(self):
+        no_floor = self._base_detector().detect_duplicates(self.entities)
+        zero_floor = self._base_detector(min_similarity=0.0).detect_duplicates(self.entities)
+        self.assertEqual(len(no_floor), len(zero_floor))
+
+    def test_min_similarity_stricter_than_threshold_reduces_results(self):
+        """A min_similarity above similarity_threshold must not increase the result count."""
+        base = self._base_detector().detect_duplicates(self.entities)
+        stricter = self._base_detector(min_similarity=0.8).detect_duplicates(self.entities)
+        self.assertLessEqual(len(stricter), len(base))
+
+    def test_min_similarity_empty_input(self):
+        detector = self._base_detector(min_similarity=0.5)
+        self.assertEqual(detector.detect_duplicates([]), [])
+
+    # ------------------------------------------------------------------
+    # sort_by
+    # ------------------------------------------------------------------
+
+    def test_sort_by_confidence_descending(self):
+        results = self._base_detector(sort_by="confidence").detect_duplicates(self.entities)
+        scores = [c.confidence for c in results]
+        self.assertEqual(scores, sorted(scores, reverse=True))
+
+    def test_sort_by_similarity_score_descending(self):
+        results = self._base_detector(sort_by="similarity_score").detect_duplicates(self.entities)
+        scores = [c.similarity_score for c in results]
+        self.assertEqual(scores, sorted(scores, reverse=True))
+
+    def test_sort_by_default_is_confidence(self):
+        default = self._base_detector().detect_duplicates(self.entities)
+        explicit = self._base_detector(sort_by="confidence").detect_duplicates(self.entities)
+        self.assertEqual(
+            [(c.entity1["id"], c.entity2["id"]) for c in default],
+            [(c.entity1["id"], c.entity2["id"]) for c in explicit],
+        )
+
+    def test_sort_by_invalid_raises_at_construction(self):
+        with self.assertRaises(ValueError):
+            self._base_detector(sort_by="bogus_field")
+
+    def test_sort_by_invalid_message_contains_field_name(self):
+        with self.assertRaises(ValueError, msg="bogus_field") as ctx:
+            self._base_detector(sort_by="bogus_field")
+        self.assertIn("bogus_field", str(ctx.exception))
+
+    # ------------------------------------------------------------------
+    # Combined options
+    # ------------------------------------------------------------------
+
+    def test_max_results_and_sort_by_similarity(self):
+        n = 2
+        results = self._base_detector(max_results=n, sort_by="similarity_score").detect_duplicates(self.entities)
+        self.assertLessEqual(len(results), n)
+        if len(results) == 2:
+            self.assertGreaterEqual(results[0].similarity_score, results[1].similarity_score)
+
+    def test_min_similarity_and_top_k_combined(self):
+        floor, k = 0.5, 1
+        results = self._base_detector(min_similarity=floor, top_k_per_entity=k).detect_duplicates(self.entities)
+        for c in results:
+            self.assertGreaterEqual(c.similarity_score, floor)
+        counts: Dict[str, int] = {}
+        for c in results:
+            for eid in (c.entity1["id"], c.entity2["id"]):
+                counts[eid] = counts.get(eid, 0) + 1
+        for count in counts.values():
+            self.assertLessEqual(count, k)
+
+    def test_all_four_options_combined(self):
+        results = self._base_detector(
+            max_results=3,
+            top_k_per_entity=1,
+            min_similarity=0.3,
+            sort_by="similarity_score",
+        ).detect_duplicates(self.entities)
+        self.assertLessEqual(len(results), 3)
+        scores = [c.similarity_score for c in results]
+        self.assertEqual(scores, sorted(scores, reverse=True))
+        for c in results:
+            self.assertGreaterEqual(c.similarity_score, 0.3)
+
+    def test_max_results_applied_after_top_k(self):
+        """max_results must slice the already-top-k-filtered list, not pre-empt it."""
+        top_k_only = self._base_detector(top_k_per_entity=1).detect_duplicates(self.entities)
+        both = self._base_detector(top_k_per_entity=1, max_results=1).detect_duplicates(self.entities)
+        self.assertLessEqual(len(both), min(1, len(top_k_only)))
+
+    # ------------------------------------------------------------------
+    # incremental_detect
+    # ------------------------------------------------------------------
+
+    def test_incremental_detect_max_results(self):
+        new_e, existing = self.entities[:3], self.entities[3:]
+        results = self._base_detector(max_results=1).incremental_detect(new_e, existing)
+        self.assertLessEqual(len(results), 1)
+
+    def test_incremental_detect_min_similarity(self):
+        new_e, existing = self.entities[:3], self.entities[3:]
+        results = self._base_detector(min_similarity=0.99).incremental_detect(new_e, existing)
+        for c in results:
+            self.assertGreaterEqual(c.similarity_score, 0.99)
+
+    def test_incremental_detect_sort_by_similarity(self):
+        new_e, existing = self.entities[:3], self.entities[3:]
+        results = self._base_detector(sort_by="similarity_score").incremental_detect(new_e, existing)
+        scores = [c.similarity_score for c in results]
+        self.assertEqual(scores, sorted(scores, reverse=True))
+
+    def test_incremental_detect_top_k_per_entity(self):
+        new_e, existing = self.entities[:3], self.entities[3:]
+        k = 1
+        results = self._base_detector(top_k_per_entity=k).incremental_detect(new_e, existing)
+        counts: Dict[str, int] = {}
+        for c in results:
+            for eid in (c.entity1["id"], c.entity2["id"]):
+                counts[eid] = counts.get(eid, 0) + 1
+        for count in counts.values():
+            self.assertLessEqual(count, k)
+
+    def test_incremental_detect_empty_new_entities(self):
+        detector = self._base_detector(max_results=5)
+        self.assertEqual(detector.incremental_detect([], self.entities), [])
+
+    def test_incremental_detect_empty_existing_entities(self):
+        detector = self._base_detector(max_results=5)
+        self.assertEqual(detector.incremental_detect(self.entities, []), [])
+
+
 if __name__ == "__main__":
     unittest.main()