diff --git a/aidrin/file_handling/readers/hdf5_reader.py b/aidrin/file_handling/readers/hdf5_reader.py
index 2d56a6e2..3d8fdf9b 100644
--- a/aidrin/file_handling/readers/hdf5_reader.py
+++ b/aidrin/file_handling/readers/hdf5_reader.py
@@ -1,8 +1,9 @@
 import os
 import uuid
-
+import json
 import h5py
 import pandas as pd
+import numpy as np
 from flask import current_app, session
 
 from aidrin.file_handling.readers.base_reader import BaseFileReader
@@ -11,134 +12,150 @@
 class hdf5Reader(BaseFileReader):
     def read(self):
         try:
-            rows = []
-            # Clean up byte strings in all object columns
-
-            def decode_bytes(df):
-                for col in df.columns:
-                    if df[col].dtype == object:
-                        df[col] = df[col].apply(
-                            lambda x: x.decode("utf-8") if isinstance(x, bytes) else x
-                        )
+            CHUNK = 10_000
+
+            MAX_TOTAL_ROWS = None
+            MAX_ROWS_PER_DATASET = None
+
+            dfs = []
+            total_rows = 0
+
+            def decode_bytes_inplace(df):
+                if df is None or df.empty:
+                    return df
+                obj_cols = df.select_dtypes(include=["object"]).columns
+                for col in obj_cols:
+                    df[col] = df[col].map(
+                        lambda x: x.decode("utf-8")
+                        if isinstance(x, (bytes, bytearray, np.bytes_))
+                        else x
+                    )
                 return df
 
-            def convert_numpy_types(obj):
-                """Recursively convert numpy types to Python native types"""
-                try:
-                    if hasattr(obj, 'item'):  # numpy scalar
-                        return obj.item()
-                    elif isinstance(obj, (list, tuple)):
-                        return [convert_numpy_types(item) for item in obj]
-                    elif isinstance(obj, dict):
-                        return {str(k): convert_numpy_types(v) for k, v in obj.items()}
-                    elif hasattr(obj, 'dtype'):  # numpy array
-                        if obj.size == 1:  # Single element array
-                            return obj.item()
-                        else:  # Multi-element array
-                            return obj.tolist()
-                    else:
-                        return obj
-                except Exception as e:
-                    self.logger.warning(f"Error converting numpy type: {e}")
-                    return str(obj)  # Fallback to string representation
-
-            def recurse(name, obj, path=[]):
-                try:
-                    if isinstance(obj, h5py.Dataset):
-                        data = obj[()]
-                        # If it's a 1D or structured dataset, load it into dicts
-                        if isinstance(data, (list, tuple)) or hasattr(data, "dtype"):
-                            try:
-                                df = pd.DataFrame(data)
-                            except Exception:
-                                df = pd.DataFrame(data.tolist())  # base
-                            for _, row in df.iterrows():
-                                try:
-                                    row_dict = row.to_dict()
-                                    # Convert any numpy types to Python native types
-                                    row_dict = convert_numpy_types(row_dict)
-                                    rows.append(row_dict)
-                                except Exception as e:
-                                    self.logger.warning(f"Error processing row: {e}")
-                                    # Try to process the row with basic conversion
-                                    try:
-                                        basic_row = {}
-                                        for col in row.index:
-                                            try:
-                                                value = row[col]
-                                                if hasattr(value, 'item'):
-                                                    basic_row[str(col)] = value.item()
-                                                else:
-                                                    basic_row[str(col)] = str(value)
-                                            except Exception:
-                                                basic_row[str(col)] = str(value)
-                                        rows.append(basic_row)
-                                    except Exception as e2:
-                                        self.logger.warning(f"Failed to process row even with basic conversion: {e2}")
-                                        continue
-                        else:
-                            # Scalar or flat dataset - ensure data is hashable
-                            try:
-                                # Convert any numpy types to Python native types
-                                data = convert_numpy_types(data)
-                                row_dict = {"value": data}
-                                rows.append(row_dict)
-                            except Exception as e:
-                                self.logger.warning(f"Error processing scalar data: {e}")
-                                # Try basic conversion
-                                try:
-                                    if hasattr(data, 'item'):
-                                        row_dict = {"value": data.item()}
-                                    else:
-                                        row_dict = {"value": str(data)}
-                                    rows.append(row_dict)
-                                except Exception as e2:
-                                    self.logger.warning(f"Failed to process scalar data even with basic conversion: {e2}")
-                                    # Skip this data point
-                                    pass
-                except Exception as e:
-                    self.logger.warning(f"Error in recurse function: {e}")
+            def make_cells_hashable_inplace(df):
+                if df is None or df.empty:
+                    return df
+
+                def to_hashable(x):
+                    if isinstance(x, (list, dict, set, tuple, np.ndarray)):
+                        try:
+                            return json.dumps(x, default=str, ensure_ascii=False)
+                        except Exception:
+                            return str(x)
+                    return x
+
+                obj_cols = df.select_dtypes(include=["object"]).columns
+                for col in obj_cols:
+                    df[col] = df[col].map(to_hashable)
+                return df
+
+            def df_from_any(data):
+                if isinstance(data, np.ndarray):
+                    if getattr(data.dtype, "names", None):
+                        return pd.DataFrame.from_records(data)
+
+                    if data.ndim == 0:
+                        return pd.DataFrame(
+                            [{"value": data.item() if isinstance(data, np.generic) else data}]
+                        )
+
+                    if data.ndim == 1:
+                        return pd.DataFrame({"value": data})
+
+                    if data.ndim > 2:
+                        data = np.ascontiguousarray(data).reshape(data.shape[0], -1)
+
+                    if data.ndim == 2:
+                        return pd.DataFrame(data)
+
+                    return pd.DataFrame({"value": [str(data)]})
+
+                if isinstance(data, (list, tuple)):
+                    try:
+                        return pd.DataFrame(data)
+                    except Exception:
+                        return pd.DataFrame({"value": list(data)})
+
+                return pd.DataFrame([{"value": data}])
+
+            def add_df(df):
+                nonlocal total_rows
+                if df is None or df.empty:
                     return
 
-            with h5py.File(self.file_path, "r") as f:
+                df.columns = [str(c) for c in df.columns]
+
+                df = decode_bytes_inplace(df)
+                df = make_cells_hashable_inplace(df)
+
+                dfs.append(df)
+                total_rows += len(df)
 
-                def visit(name, obj):
-                    recurse(name, obj, name.strip("/").split("/"))
+            with h5py.File(self.file_path, "r") as f:
 
-                f.visititems(visit)
-            df = pd.DataFrame(rows)
-            df = decode_bytes(df)
+                def handle_dataset(name, dset):
+                    nonlocal total_rows
+                    try:
+                        shape = getattr(dset, "shape", None)
+                        dtype = getattr(dset, "dtype", None)
+                        self.logger.info(f"HDF5 dataset: {name} shape={shape} dtype={dtype}")
+
+                        if shape is None or len(shape) == 0:
+                            df = df_from_any(dset[()])
+                            add_df(df)
+                            return
+
+                        n = int(shape[0]) if shape[0] is not None else 0
+                        if n == 0:
+                            df = df_from_any(dset[()])
+                            add_df(df)
+                            return
+
+                        limit = n
+                        if isinstance(MAX_ROWS_PER_DATASET, int):
+                            limit = min(limit, MAX_ROWS_PER_DATASET)
+
+                        for start in range(0, limit, CHUNK):
+                            if isinstance(MAX_TOTAL_ROWS, int) and total_rows >= MAX_TOTAL_ROWS:
+                                self.logger.warning("Stopping early due to MAX_TOTAL_ROWS cap")
+                                return
+
+                            chunk = dset[start: start + CHUNK]
+                            df = df_from_any(chunk)
+                            add_df(df)
+
+                    except Exception:
+                        self.logger.exception(f"Failed reading dataset: {name}")
+
+                def visitor(name, obj):
+                    if isinstance(obj, h5py.Dataset):
+                        handle_dataset(name, obj)
 
-            # Ensure all column names are strings to avoid numpy array issues
-            if hasattr(df, 'columns') and len(df.columns) > 0:
-                df.columns = [str(col) for col in df.columns]
+                f.visititems(visitor)
 
-            # Check if DataFrame is empty and log warning
-            if df.empty:
+            if not dfs:
                 self.logger.warning("No data was successfully processed from HDF5 file")
                 return None
 
-            return df
+            out = pd.concat(dfs, ignore_index=True, sort=False)
+            return None if out.empty else out
+
         except Exception as e:
             self.logger.error(f"Error while reading: {e}")
+            self.logger.exception("HDF5 read() failed with exception")
             return None
 
     def parse(self):
-        # Recursively find all group names in the HDF5 file
         def recurse(data):
             try:
-                # Convert items() to a list first to avoid iteration issues
                 items = list(data.items())
                 for name, obj in items:
                     try:
-                        # Ensure name is a string and hashable to avoid "unhashable type" errors
                         full_path = str(name)
-
                         if isinstance(obj, h5py.Group):
                             group_names.append(full_path)
                             recurse(obj)
                     except (TypeError, ValueError) as e:
-                        # If conversion fails, skip this key and log the error
                         self.logger.warning(f"Skipping unhashable key {name}: {e}")
                         continue
             except Exception as e:
@@ -154,23 +171,19 @@ def recurse(data):
     def filter(self, kept_keys):
         if isinstance(kept_keys, str):
             kept_keys = kept_keys.split(",")
-        # Ensure all keys are strings and hashable to avoid "unhashable type" errors
+
         filtered_keys = set()
         for g in kept_keys:
             try:
-                # Convert to string and ensure it's hashable
                 key_str = str(g).strip("/")
-                # Test if it's hashable by trying to add to set
                 filtered_keys.add(key_str)
             except (TypeError, ValueError) as e:
-                # If conversion fails, skip this key and log the error
                 self.logger.warning(f"Skipping unhashable key {g}: {e}")
                 continue
 
-        new_file_name = (
-            f"filtered_{uuid.uuid4().hex}_{session.get('uploaded_file_name')}"
-        )
+        new_file_name = f"filtered_{uuid.uuid4().hex}_{session.get('uploaded_file_name')}"
         new_file_path = os.path.join(current_app.config["UPLOAD_FOLDER"], new_file_name)
+
         with (
             h5py.File(self.file_path, "r") as src,
             h5py.File(new_file_path, "w") as tgt,
@@ -179,12 +192,14 @@ def filter(self, kept_keys):
             def copy_group(path, src_group, tgt_group):
                 for name, obj in src_group.items():
                     full_path = f"{path}/{name}".strip("/")
+
                     if isinstance(obj, h5py.Group):
                         if full_path in filtered_keys:
                             tgt_subgroup = tgt_group.create_group(name)
                             copy_group(full_path, obj, tgt_subgroup)
                         else:
                             copy_group(full_path, obj, tgt_group)
+
                     elif isinstance(obj, h5py.Dataset):
                         if path.strip("/") in filtered_keys:
                             tgt_group.create_dataset(name, data=obj[()])
@@ -192,3 +207,5 @@ def copy_group(path, src_group, tgt_group):
             copy_group("", src, tgt)
 
         return new_file_path
+
+
diff --git a/aidrin/structured_data_metrics/completeness.py b/aidrin/structured_data_metrics/completeness.py
index 9e284534..9c026521 100644
--- a/aidrin/structured_data_metrics/completeness.py
+++ b/aidrin/structured_data_metrics/completeness.py
@@ -1,70 +1,210 @@
 import base64
 import io
+from typing import Optional
 
+import matplotlib
+
+matplotlib.use("Agg")  
 import matplotlib.pyplot as plt
+import numpy as np
 from celery import Task, shared_task
 from celery.exceptions import SoftTimeLimitExceeded
 
 from aidrin.file_handling.file_parser import read_file
 
 
+def iterate_chunks(df, chunksize: int = 50_000):
+    for start in range(0, len(df), chunksize):
+        yield df.iloc[start : start + chunksize]
+
+
+MAX_B64_LEN = 250_000
+MAX_BARS = 40
+CHUNK_SIZE = 50_000
+MAX_CHUNK_STATS = 200  # cap response size
+
+
+def _fig_to_b64(dpi: int = 85) -> str:
+    buf = io.BytesIO()
+    plt.savefig(buf, format="png", dpi=dpi, bbox_inches="tight")
+    buf.seek(0)
+    s = base64.b64encode(buf.read()).decode("utf-8")
+    plt.close()
+    return s
+
+
+def _info_image(message: str) -> str:
+    plt.figure(figsize=(7.5, 2.3))
+    plt.axis("off")
+    plt.text(0.01, 0.5, message, fontsize=11, va="center")
+    plt.tight_layout()
+    return _fig_to_b64(dpi=95)
+
+
+def _barplot_small(
+    keys,
+    vals,
+    title: str,
+    max_bars: int = MAX_BARS,
+    worst: bool = True,
+) -> Optional[str]:
+    keys = list(keys)
+    vals = np.asarray(list(vals), dtype=float)
+
+    mask = np.isfinite(vals)
+    keys = [k for k, m in zip(keys, mask) if m]
+    vals = vals[mask]
+    if vals.size == 0:
+        return None
+
+    if len(keys) > max_bars:
+        order = np.argsort(vals)
+        order = order[:max_bars] if worst else order[-max_bars:]
+        keys = [keys[i] for i in order]
+        vals = vals[order]
+
+    x = np.arange(len(keys))
+    plt.figure(figsize=(7.5, 3.2))
+    plt.bar(x, vals)
+    plt.title(title, fontsize=12)
+    plt.ylim(0, 1)
+
+    step = max(1, len(keys) // 10)
+    tick_idx = np.arange(0, len(keys), step)
+    plt.xticks(
+        tick_idx,
+        [str(keys[i]) for i in tick_idx],
+        rotation=45,
+        ha="right",
+        fontsize=8,
+    )
+
+    plt.tight_layout()
+    return _fig_to_b64(dpi=85)
+
+
 @shared_task(bind=True, ignore_result=False)
 def completeness(self: Task, file_info):
     try:
-        file = read_file(file_info)
-
-        # Ensure DataFrame columns are strings to avoid numpy array issues
-        if hasattr(file, 'columns'):
-            file.columns = [str(col) for col in file.columns]
-
-        # Calculate completeness metric for each column
-        try:
-            completeness_scores = (1 - file.isnull().mean()).to_dict()
-        except Exception:
-            # If to_dict() fails, manually create the dictionary
-            completeness_scores = {}
-            for col in file.columns:
-                try:
-                    completeness_scores[str(col)] = float(1 - file[col].isnull().mean())
-                except Exception:
-                    completeness_scores[str(col)] = 0.0
-
-        # Ensure all column names are strings to avoid numpy array issues
-        completeness_scores = {str(k): v for k, v in completeness_scores.items()}
-
-        # Calculate overall completeness metric for the dataset
-        overall_completeness = 1 - file.isnull().any(axis=1).mean()
-
-        result_dict = {}
-
-        # Always include completeness scores for all features
-        result_dict["Completeness scores"] = completeness_scores
-        result_dict["Overall Completeness"] = overall_completeness
-
-        # Create a bar chart for all features
-        plt.figure(figsize=(8, 6))
-        plt.bar(completeness_scores.keys(), completeness_scores.values(), color="blue")
-        plt.title("Feature-wise Completeness Scores", fontsize=16)
-        plt.xlabel("Features", fontsize=14)
-        plt.ylabel("Completeness Score", fontsize=14)
-        plt.ylim(0, 1)
-
-        # Rotate x-axis tick labels for readability
-        plt.xticks(rotation=45, ha="right", fontsize=12)
-        plt.tight_layout()
-
-        # Save the chart to a BytesIO object
-        img_buf = io.BytesIO()
-        plt.savefig(img_buf, format="png")
-        img_buf.seek(0)
-
-        # Encode the image as base64
-        img_base64 = base64.b64encode(img_buf.read()).decode("utf-8")
-
-        result_dict["Completeness Visualization"] = img_base64
-        plt.close()
-
-        return result_dict
+        df = read_file(file_info)
+
+        if df is None:
+            return {
+                "Error": "File could not be read.",
+                "Completeness Visualization": _info_image(
+                    "Completeness unavailable: file could not be read."
+                ),
+            }
+
+        if not hasattr(df, "columns") or df.empty:
+            return {
+                "Error": "Dataset is empty.",
+                "Completeness Visualization": _info_image(
+                    "Completeness unavailable: dataset is empty."
+                ),
+            }
+
+        # Ensure column names are plain strings
+        df.columns = [str(c) for c in df.columns]
+        cols = list(df.columns)
+
+        total_rows = 0
+        total_missing_per_col = {col: 0 for col in cols}
+        total_rows_with_any_missing = 0
+
+        chunk_stats = []
+
+        for idx, chunk in enumerate(iterate_chunks(df, chunksize=CHUNK_SIZE)):
+            chunk_size = int(len(chunk))
+            if chunk_size == 0:
+                continue
+
+            total_rows += chunk_size
+
+            miss = chunk.isna().sum().to_dict()
+            for col in cols:
+                total_missing_per_col[col] += int(miss.get(col, 0))
+
+            rows_any_missing = int(chunk.isna().any(axis=1).sum())
+            total_rows_with_any_missing += rows_any_missing
+
+            
+            if idx < MAX_CHUNK_STATS:
+                chunk_overall = float(1.0 - (rows_any_missing / chunk_size))
+                chunk_stats.append(
+                    {
+                        "chunk": idx,
+                        "size": chunk_size,
+                        "overall": chunk_overall,
+                    }
+                )
+
+        if total_rows == 0:
+            return {
+                "Error": "No rows found in dataset.",
+                "Completeness Visualization": _info_image(
+                    "Completeness unavailable: no rows found."
+                ),
+            }
+
+        # Final feature-wise completeness
+        final_feature_scores = {
+            col: float(1.0 - (total_missing_per_col[col] / total_rows)) for col in cols
+        }
+        # Final overall completeness (row has no missing values)
+        final_overall = float(1.0 - (total_rows_with_any_missing / total_rows))
+
+        out = {
+            
+            "Completeness scores": final_feature_scores,
+            "Overall Completeness": final_overall,
+            
+            "Chunk Completeness": chunk_stats,
+            "Final Completeness": {
+                "feature_wise": final_feature_scores,
+                "overall": final_overall,
+            },
+        }
+
+        vals = np.asarray(list(final_feature_scores.values()), dtype=float)
+        finite_vals = vals[np.isfinite(vals)]
+
+        if finite_vals.size == 0:
+            out["Completeness Visualization"] = _info_image(
+                "Completeness computed, but no finite scores to plot."
+            )
+            return out
+
+        if float(np.nanmin(finite_vals)) == 1.0 and float(np.nanmax(finite_vals)) == 1.0:
+            out["Completeness Visualization"] = _info_image(
+                "All features are 100% complete.\nNo missing values detected."
+            )
+            return out
+
+        img = _barplot_small(
+            keys=list(final_feature_scores.keys()),
+            vals=list(final_feature_scores.values()),
+            title="Feature-wise Completeness (worst features)",
+            max_bars=MAX_BARS,
+            worst=True,
+        )
+
+        if (not img) or (len(img) > MAX_B64_LEN):
+            out["Completeness Visualization"] = _info_image(
+                "Completeness chart omitted to prevent oversized payload.\n"
+                "Scores are computed successfully."
+            )
+        else:
+            out["Completeness Visualization"] = img
+
+        return out
 
     except SoftTimeLimitExceeded:
         raise Exception("Completeness task timed out.")
+    except Exception as e:
+        msg = f"Completeness calculation failed: {str(e)}"
+        return {
+            "Error": msg,
+            "Completeness Visualization": _info_image(f"{msg}"),
+        }
+
diff --git a/aidrin/structured_data_metrics/duplicity.py b/aidrin/structured_data_metrics/duplicity.py
index c43701c6..72d16b5f 100644
--- a/aidrin/structured_data_metrics/duplicity.py
+++ b/aidrin/structured_data_metrics/duplicity.py
@@ -1,5 +1,7 @@
 from celery import Task, shared_task
 from celery.exceptions import SoftTimeLimitExceeded
+import numpy as np
+import pandas as pd
 
 from aidrin.file_handling.file_parser import read_file
 
@@ -8,14 +10,59 @@
 def duplicity(self: Task, file_info):
     try:
         file = read_file(file_info)
-        dup_dict = {}
-        # Calculate the proportion of duplicate values
-        duplicate_proportions = file.duplicated().sum() / len(file)
 
-        dup_dict["Duplicity scores"] = {
-            "Overall duplicity of the dataset": duplicate_proportions
+        if file is None or not hasattr(file, "empty") or file.empty:
+            return {
+                "Duplicity scores": {
+                    "Overall duplicity of the dataset": 0.0
+                }
+            }
+
+        n_rows = len(file)
+        n_cols = len(file.columns)
+
+        if n_rows >= 2:
+            dup = float(file.duplicated().sum() / n_rows)
+            return {
+                "Duplicity scores": {
+                    "Overall duplicity of the dataset": dup
+                }
+            }
+
+        numeric = file.select_dtypes(include=[np.number])
+
+        if numeric.empty:
+            vals = file.to_numpy().ravel()
+            vals = vals[~pd.isna(vals)]
+            total = int(vals.size)
+            if total <= 1:
+                dup = 0.0
+            else:
+                uniq = int(pd.unique(vals).size)
+                dup = float(1.0 - (uniq / total))
+            return {
+                "Duplicity scores": {
+                    "Overall duplicity of the dataset": dup
+                }
+            }
+
+        vals = numeric.to_numpy().ravel()
+        vals = vals[np.isfinite(vals)]
+        total = int(vals.size)
+
+        if total <= 1:
+            dup = 0.0
+        else:
+            vals = np.round(vals, 6)
+            uniq = int(pd.unique(vals).size)
+            dup = float(1.0 - (uniq / total))
+
+        return {
+            "Duplicity scores": {
+                "Overall duplicity of the dataset": dup
+            }
         }
 
-        return dup_dict
     except SoftTimeLimitExceeded:
         raise Exception("Duplicity task timed out.")
+
diff --git a/aidrin/structured_data_metrics/outliers.py b/aidrin/structured_data_metrics/outliers.py
index 945ac31e..aa70e3e3 100644
--- a/aidrin/structured_data_metrics/outliers.py
+++ b/aidrin/structured_data_metrics/outliers.py
@@ -1,97 +1,207 @@
 import base64
 import io
+from typing import Optional
 
+import matplotlib
+
+matplotlib.use("Agg")  
 import matplotlib.pyplot as plt
 import numpy as np
+import pandas as pd
 from celery import Task, shared_task
 from celery.exceptions import SoftTimeLimitExceeded
 
 from aidrin.file_handling.file_parser import read_file
 
 
-@shared_task(bind=True, ignore_result=False)
-def outliers(self: Task, file_info):
-    try:
-        file = read_file(file_info)
-
-        # Ensure DataFrame columns are strings to avoid numpy array issues
-        if hasattr(file, 'columns'):
-            file.columns = [str(col) for col in file.columns]
-
-        try:
-            out_dict = {}
-            # Select numerical columns for outlier detection
-            numerical_columns = file.select_dtypes(include=[np.number])
-
-            if numerical_columns.empty:
-                return {"Error": "No numerical features found in the dataset."}
-
-            proportions_dict = {}
-
-            # Process each column separately
-            for col in numerical_columns.columns:
-                series = numerical_columns[col].dropna()
-
-                if series.empty:
-                    proportions_dict[col] = np.nan
-                    continue
-
-                q1 = series.quantile(0.25)
-                q3 = series.quantile(0.75)
-                IQR = q3 - q1
-
-                if IQR == 0:
-                    proportions_dict[col] = 0.0  # no variability, no outliers
-                    continue
+MAX_B64_LEN = 250_000
+MAX_BARS = 40
+
+
+def _fig_to_b64(fig: plt.Figure, dpi: int = 85) -> str:
+    """Convert a specific Matplotlib Figure to base64 PNG."""
+    buf = io.BytesIO()
+    fig.savefig(buf, format="png", dpi=dpi, bbox_inches="tight")
+    buf.seek(0)
+    s = base64.b64encode(buf.read()).decode("utf-8")
+    plt.close(fig)
+    return s
+
+
+def _info_image(message: str) -> str:
+    fig = plt.figure(figsize=(7.5, 2.3))
+    ax = fig.add_subplot(111)
+    ax.axis("off")
+    ax.text(0.01, 0.5, message, fontsize=11, va="center")
+    fig.tight_layout()
+    return _fig_to_b64(fig, dpi=95)
+
+
+def _barplot_small(
+    keys,
+    vals,
+    title: str,
+    max_bars: int = MAX_BARS,
+    top: bool = True,
+) -> Optional[str]:
+    keys = list(keys)
+    vals = np.asarray(list(vals), dtype=float)
+
+    mask = np.isfinite(vals)
+    keys = [k for k, m in zip(keys, mask) if m]
+    vals = vals[mask]
+    if vals.size == 0:
+        return None
+
+    if len(keys) > max_bars:
+        order = np.argsort(vals)
+        order = order[-max_bars:] if top else order[:max_bars]
+        keys = [keys[i] for i in order]
+        vals = vals[order]
+
+    x = np.arange(len(keys))
+    fig = plt.figure(figsize=(7.5, 3.2))
+    ax = fig.add_subplot(111)
+    ax.bar(x, vals)
+    ax.set_title(title, fontsize=12)
+    ax.set_ylim(0, 1)
+
+    step = max(1, len(keys) // 10)
+    tick_idx = np.arange(0, len(keys), step)
+    ax.set_xticks(tick_idx)
+    ax.set_xticklabels(
+        [str(keys[i]) for i in tick_idx],
+        rotation=45,
+        ha="right",
+        fontsize=8,
+    )
+
+    fig.tight_layout()
+    return _fig_to_b64(fig, dpi=85)
 
-                # Identify outliers using IQR
-                mask = (series < (q1 - 1.5 * IQR)) | (series > (q3 + 1.5 * IQR))
-                proportions_dict[col] = mask.mean()  # proportion of outliers
 
-            # Calculate overall outlier score
-            valid_values = [v for v in proportions_dict.values() if not np.isnan(v)]
-            overall_score = np.mean(valid_values) if valid_values else 0.0
-            proportions_dict["Overall outlier score"] = overall_score
-
-            # Ensure all column names are strings to avoid numpy array issues
-            proportions_dict = {str(k): v for k, v in proportions_dict.items()}
-
-            # Calculate the average of dictionary values
-            average_value = sum(proportions_dict.values()) / len(proportions_dict)
-            proportions_dict["Overall outlier score"] = average_value
-            # add the average to dictionary
-            out_dict["Outlier scores"] = proportions_dict
+@shared_task(bind=True, ignore_result=False)
+def outliers(self: Task, file_info):
 
-            # Create bar chart for feature-level outlier proportions only
-            feature_scores = {
-                k: v for k, v in proportions_dict.items() if k != "Overall outlier score"
+    try:
+        df = read_file(file_info)
+
+        if df is None:
+            return {
+                "Error": "File could not be read.",
+                "Outliers Visualization": _info_image(
+                    "Outliers unavailable: file could not be read."
+                ),
             }
 
-            if feature_scores:  # only plot if there are valid features
-                plt.figure(figsize=(8, 8))
-                plt.bar(feature_scores.keys(), feature_scores.values(), color="red")
-                plt.title("Proportion of Outliers for Numerical Columns", fontsize=14)
-                plt.xlabel("Columns", fontsize=14)
-                plt.ylabel("Proportion of Outliers", fontsize=14)
-                plt.ylim(0, 1)
-
-                plt.xticks(rotation=45, ha="right", fontsize=12)
-                plt.subplots_adjust(bottom=0.5)
-                plt.tight_layout()
-
-                # Save the chart to BytesIO and encode as base64
-                img_buf = io.BytesIO()
-                plt.savefig(img_buf, format="png")
-                img_buf.seek(0)
-                img_base64 = base64.b64encode(img_buf.read()).decode("utf-8")
-
-                out_dict["Outliers Visualization"] = img_base64
-                plt.close()
+        if not hasattr(df, "columns") or df.empty:
+            return {
+                "Error": "Dataset is empty.",
+                "Outliers Visualization": _info_image(
+                    "Outliers unavailable: dataset is empty."
+                ),
+            }
 
-            return out_dict
+        df.columns = [str(c) for c in df.columns]
+
+
+        for c in df.columns:
+            if df[c].dtype == object:
+                coerced = pd.to_numeric(df[c], errors="coerce")
+       
+                if np.isfinite(coerced.to_numpy(dtype=float, copy=False)).sum() > 0:
+                    df[c] = coerced
+
+        numerical_df = df.select_dtypes(include=[np.number])
+        if numerical_df.empty:
+ 
+            return {
+                "Error": "No numerical features found in the dataset.",
+                "Outlier scores": {"Overall outlier score": 0.0},
+                "Outliers Visualization": _info_image(
+                    "No numerical features found in the dataset."
+                ),
+            }
 
-        except Exception as e:
-            return {"Error": f"Outlier detection failed: {str(e)}"}
+        proportions = {}
+
+        for col in numerical_df.columns:
+            series = numerical_df[col].dropna()
+
+            if series.empty:
+                proportions[str(col)] = np.nan
+                continue
+
+            q1 = float(series.quantile(0.25))
+            q3 = float(series.quantile(0.75))
+            iqr = float(q3 - q1)
+
+            if not np.isfinite(iqr) or iqr == 0.0:
+                proportions[str(col)] = 0.0
+                continue
+
+            lower = q1 - 1.5 * iqr
+            upper = q3 + 1.5 * iqr
+            mask = (series < lower) | (series > upper)
+            proportions[str(col)] = float(mask.mean())
+
+        valid_values = [v for v in proportions.values() if np.isfinite(v)]
+        overall_score = float(np.mean(valid_values)) if valid_values else 0.0
+        proportions["Overall outlier score"] = overall_score
+
+        out = {"Outlier scores": proportions}
+
+        # Visualization (feature-level only)
+        feature_scores = {
+            k: v for k, v in proportions.items() if k != "Overall outlier score"
+        }
+
+        if not feature_scores:
+            out["Outliers Visualization"] = _info_image(
+                "Outliers computed, but no feature scores to plot."
+            )
+            return out
+
+        vals = np.asarray(list(feature_scores.values()), dtype=float)
+        finite_vals = vals[np.isfinite(vals)]
+
+        if finite_vals.size == 0:
+            out["Outliers Visualization"] = _info_image(
+                "Outliers computed, but no finite scores to plot."
+            )
+            return out
+
+        if float(np.nanmax(finite_vals)) == 0.0:
+            out["Outliers Visualization"] = _info_image(
+                "No outliers detected (all proportions are 0).\n"
+                "Common causes: very few rows, constant columns (IQR=0)."
+            )
+            return out
+
+        img = _barplot_small(
+            keys=list(feature_scores.keys()),
+            vals=list(feature_scores.values()),
+            title="Proportion of Outliers (top features)",
+            max_bars=MAX_BARS,
+            top=True,
+        )
+
+        if (not img) or (len(img) > MAX_B64_LEN):
+            out["Outliers Visualization"] = _info_image(
+                "Outliers chart omitted to prevent oversized payload.\n"
+                "Scores are computed successfully."
+            )
+        else:
+            out["Outliers Visualization"] = img
+
+        return out
 
     except SoftTimeLimitExceeded:
         raise Exception("Outliers task timed out.")
+    except Exception as e:
+        msg = f"Outlier detection failed: {str(e)}"
+        return {
+            "Error": msg,
+            "Outliers Visualization": _info_image(msg),
+        }
+