diff --git a/litellm/caching/in_memory_cache.py b/litellm/caching/in_memory_cache.py
index 1cc3e106aa0..6b40c48c4ef 100644
--- a/litellm/caching/in_memory_cache.py
+++ b/litellm/caching/in_memory_cache.py
@@ -274,7 +274,9 @@ class InMemoryCache(BaseCache):
     ) -> list[float] | None:
         results: Final = []
         for increment in increment_list:
-            result = await self.async_increment(increment["key"], increment["increment_value"], **kwargs)
+            result = await self.async_increment(
+                increment["key"], increment["increment_value"], ttl=increment["ttl"], **kwargs
+            )
             results.append(result)
         return results
 
diff --git a/litellm/litellm_core_utils/get_litellm_params.py b/litellm/litellm_core_utils/get_litellm_params.py
index ba964d200ec..8178b7e98b8 100644
--- a/litellm/litellm_core_utils/get_litellm_params.py
+++ b/litellm/litellm_core_utils/get_litellm_params.py
@@ -75,6 +75,8 @@ OPTIONAL_KWARGS_KEYS: Final = (
             "rpm",
             "itpm",
             "otpm",
+            "tpd",
+            "rpd",
             "use_xai_oauth",
             "chatgpt_auth_file",
             "github_copilot_auth_type",
diff --git a/litellm/router_utils/pre_call_checks/model_rate_limit_check.py b/litellm/router_utils/pre_call_checks/model_rate_limit_check.py
index ea3fdd601dc..9eebcf39515 100644
--- a/litellm/router_utils/pre_call_checks/model_rate_limit_check.py
+++ b/litellm/router_utils/pre_call_checks/model_rate_limit_check.py
@@ -5,15 +5,19 @@ When enabled via router_settings.optional_pre_call_checks: ["enforce_model_rate_
 
 - tpm/rpm: combined TPM + optional RPM (legacy)
 - itpm/otpm: separate input/output tokens per minute
+- tpd/rpd: tokens and requests per UTC calendar day
 
 When a deployment sets both itpm/otpm and tpm/rpm, both are enforced. A warning
 is logged the first time such a deployment is seen.
 """
 
 import contextlib
-from typing import TYPE_CHECKING, Any, Final
+from collections.abc import Callable
+from datetime import datetime, timedelta
+from typing import TYPE_CHECKING, Any, Final, Literal
 
 import httpx
+from pydantic import BaseModel, ConfigDict
 
 import litellm
 from litellm._internal_context import with_service_target
@@ -33,6 +37,7 @@ from litellm.router_utils.pre_call_checks.io_token_rate_limit_check import (
     io_token_refund_failure,
 )
 from litellm.router_utils.routing_read_batch import ROUTER_USAGE_TARGET
+from litellm.types.caching import RedisPipelineIncrementOperation
 from litellm.types.router import RouterErrors
 from litellm.types.utils import StandardLoggingPayload
 from litellm.utils import get_utc_datetime
@@ -47,6 +52,62 @@ else:
 
 class RoutingArgs:
     ttl: int = 60  # 1min (RPM/TPM expire key)
+    daily_ttl: int = 86400
+
+
+DAY_FORMAT: Final = "%Y-%m-%d"
+
+
+def seconds_until_utc_midnight(now: datetime) -> int:
+    next_midnight: Final = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
+    return max(1, int((next_midnight - now).total_seconds()))
+
+
+def _first_set(*values: int | None) -> int | None:
+    return next((value for value in values if value is not None), None)
+
+
+class _DailyLimitFields(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="ignore", from_attributes=True)
+    tpd: int | None = None
+    rpd: int | None = None
+
+
+class _DailyLimitParams(_DailyLimitFields):
+    model: str | None = None
+
+
+class _DailyLimitModelInfo(_DailyLimitFields):
+    id: str | None = None
+
+
+class _DailyLimitDeployment(_DailyLimitFields):
+    model_name: str = ""
+    litellm_params: _DailyLimitParams = _DailyLimitParams()
+    model_info: _DailyLimitModelInfo = _DailyLimitModelInfo()
+
+    def limits(self) -> tuple[int | None, int | None]:
+        return (
+            _first_set(self.tpd, self.litellm_params.tpd, self.model_info.tpd),
+            _first_set(self.rpd, self.litellm_params.rpd, self.model_info.rpd),
+        )
+
+    def cache_keys(self, current_day: str) -> tuple[str, str]:
+        prefix: Final = f"{self.model_info.id}:{self.litellm_params.model}"
+        return f"{prefix}:tpd:{current_day}", f"{prefix}:rpd:{current_day}"
+
+    def limit_error(self, kind: Literal["tpd", "rpd"], limit: int, usage: int, now: datetime) -> litellm.RateLimitError:
+        return litellm.RateLimitError(
+            message=f"Model rate limit exceeded. {kind.upper()} limit={limit}, current usage={usage}",
+            llm_provider="",
+            model=self.litellm_params.model,
+            response=httpx.Response(
+                status_code=429,
+                content=f"{RouterErrors.user_defined_ratelimit_error.value} {kind} limit={limit}. current usage={usage}. id={self.model_info.id}, model_group={self.model_name}",
+                headers={"retry-after": str(seconds_until_utc_midnight(now))},
+                request=httpx.Request(method="model_rate_limit_check", url="https://github.com/BerriAI/litellm"),
+            ),
+        )
 
 
 class ModelRateLimitingCheck(CustomLogger):
@@ -60,12 +121,60 @@ class ModelRateLimitingCheck(CustomLogger):
     this check actively enforces those limits across ALL routing strategies.
     """
 
-    def __init__(self, dual_cache: DualCache):
+    def __init__(self, dual_cache: DualCache, clock: Callable[[], datetime] = get_utc_datetime):
         self.dual_cache = dual_cache
+        self.clock = clock
         # model_ids already warned about itpm/otpm + tpm/rpm on the same deployment,
         # so the warning is logged once per deployment rather than per request.
         self._io_token_conflict_warned_ids: set[str] = set()
 
+    def _increment(self, key: str, value: int, ttl: int) -> int:
+        return self.dual_cache.increment_cache(key=key, value=value, ttl=ttl)
+
+    async def _async_increment(
+        self, key: str, value: int, ttl: int, parent_otel_span: Span | None = None
+    ) -> int | None:
+        count: Final = await self.dual_cache.async_increment_cache(
+            key=key, value=value, ttl=ttl, parent_otel_span=parent_otel_span
+        )
+        return None if count is None else int(count)
+
+    @staticmethod
+    def _request_count_operations(
+        rpm_key: str | None, rpd_key: str | None, increment_value: int
+    ) -> tuple[RedisPipelineIncrementOperation, ...]:
+        return tuple(
+            RedisPipelineIncrementOperation(key=key, increment_value=increment_value, ttl=ttl)
+            for key, ttl in ((rpm_key, RoutingArgs.ttl), (rpd_key, RoutingArgs.daily_ttl))
+            if key is not None
+        )
+
+    async def _async_refund_request_counts(
+        self, rpm_key: str | None, rpd_key: str | None, parent_otel_span: Span | None
+    ) -> None:
+        operations: Final = self._request_count_operations(rpm_key, rpd_key, -1)
+        if operations:
+            await self.dual_cache.async_increment_cache_pipeline(
+                increment_list=list(operations), parent_otel_span=parent_otel_span
+            )
+
+    async def _async_increment_request_counts(
+        self, rpm_key: str | None, rpd_key: str | None, parent_otel_span: Span | None
+    ) -> tuple[int | None, int | None]:
+        operations: Final = self._request_count_operations(rpm_key, rpd_key, 1)
+        if not operations:
+            return None, None
+        counts: Final = await self.dual_cache.async_increment_cache_pipeline(
+            increment_list=list(operations), parent_otel_span=parent_otel_span
+        )
+        if counts is None or len(counts) != len(operations):
+            return None, None
+        count_by_key: Final = {op["key"]: int(count) for op, count in zip(operations, counts)}
+        return (
+            None if rpm_key is None else count_by_key.get(rpm_key),
+            None if rpd_key is None else count_by_key.get(rpd_key),
+        )
+
     def _warn_io_token_and_tpm_rpm_coexist_once(self, deployment: dict) -> None:
         tpm_limit, rpm_limit = self._get_deployment_limits(deployment)
         if tpm_limit is None and rpm_limit is None:
@@ -129,6 +238,21 @@ class ModelRateLimitingCheck(CustomLogger):
 
         return tpm, rpm
 
+    def _check_daily_limits(self, deployment: _DailyLimitDeployment, now: datetime) -> None:
+        tpd_limit, rpd_limit = deployment.limits()
+        if tpd_limit is None and rpd_limit is None:
+            return
+        tpd_key, rpd_key = deployment.cache_keys(now.strftime(DAY_FORMAT))
+        if tpd_limit is not None:
+            current_tpd: Final = self._get_current_tpm(tpd_key, tpd_limit)
+            if current_tpd is not None and current_tpd >= tpd_limit:
+                raise deployment.limit_error("tpd", tpd_limit, current_tpd, now)
+        if rpd_limit is not None:
+            current_rpd: Final = self._increment(rpd_key, 1, RoutingArgs.daily_ttl)
+            if current_rpd > rpd_limit:
+                self._increment(rpd_key, -1, RoutingArgs.daily_ttl)
+                raise deployment.limit_error("rpd", rpd_limit, current_rpd, now)
+
     def _get_cache_keys(self, deployment: dict, current_minute: str) -> tuple[str, str]:
         """Get the cache keys for TPM and RPM tracking."""
         model_id: Final = deployment.get("model_info", {}).get("id")
@@ -178,13 +302,10 @@ class ModelRateLimitingCheck(CustomLogger):
                 )
                 io_reservation_made = True
 
+            dt: Final = self.clock()
+            daily: Final = _DailyLimitDeployment.model_validate(deployment)
             tpm_limit, rpm_limit = self._get_deployment_limits(deployment)
 
-            # If no limits are set, allow the request
-            if tpm_limit is None and rpm_limit is None:
-                return deployment
-
-            dt: Final = get_utc_datetime()
             current_minute: Final = dt.strftime("%H-%M")
             tpm_key, rpm_key = self._get_cache_keys(deployment, current_minute)
 
@@ -213,8 +334,9 @@ class ModelRateLimitingCheck(CustomLogger):
 
             # Check RPM limit (atomic increment-first to avoid race conditions)
             if rpm_limit is not None:
-                current_rpm: Final = self.dual_cache.increment_cache(key=rpm_key, value=1, ttl=RoutingArgs.ttl)
+                current_rpm: Final = self._increment(rpm_key, 1, RoutingArgs.ttl)
                 if current_rpm is not None and current_rpm > rpm_limit:
+                    self._increment(rpm_key, -1, RoutingArgs.ttl)
                     raise litellm.RateLimitError(
                         message=f"Model rate limit exceeded. RPM limit={rpm_limit}, current usage={current_rpm}",
                         llm_provider="",
@@ -230,6 +352,12 @@ class ModelRateLimitingCheck(CustomLogger):
                         ),
                     )
 
+            try:
+                self._check_daily_limits(daily, dt)
+            except litellm.RateLimitError:
+                if rpm_limit is not None:
+                    self._increment(rpm_key, -1, RoutingArgs.ttl)
+                raise
             return deployment
 
         except litellm.RateLimitError:
@@ -259,15 +387,14 @@ class ModelRateLimitingCheck(CustomLogger):
                 )
                 io_reservation_made = True
 
+            dt: Final = self.clock()
+            daily: Final = _DailyLimitDeployment.model_validate(deployment)
+            tpd_limit, rpd_limit = daily.limits()
             tpm_limit, rpm_limit = self._get_deployment_limits(deployment)
 
-            # If no limits are set, allow the request
-            if tpm_limit is None and rpm_limit is None:
-                return deployment
-
-            dt: Final = get_utc_datetime()
             current_minute: Final = dt.strftime("%H-%M")
             tpm_key, rpm_key = self._get_cache_keys(deployment, current_minute)
+            tpd_key, rpd_key = daily.cache_keys(dt.strftime(DAY_FORMAT))
 
             model_id: Final = deployment.get("model_info", {}).get("id")
             model_name: Final = deployment.get("litellm_params", {}).get("model")
@@ -293,30 +420,37 @@ class ModelRateLimitingCheck(CustomLogger):
                         num_retries=0,  # Don't retry - return 429 immediately
                     )
 
-            # Check RPM limit (atomic increment-first to avoid race conditions)
-            if rpm_limit is not None:
-                current_rpm: Final = await self.dual_cache.async_increment_cache(
-                    key=rpm_key,
-                    value=1,
-                    ttl=RoutingArgs.ttl,
-                    parent_otel_span=parent_otel_span,
-                )
-                if current_rpm is not None and current_rpm > rpm_limit:
-                    raise litellm.RateLimitError(
-                        message=f"Model rate limit exceeded. RPM limit={rpm_limit}, current usage={current_rpm}",
-                        llm_provider="",
-                        model=model_name,
-                        response=httpx.Response(
-                            status_code=429,
-                            content=f"{RouterErrors.user_defined_ratelimit_error.value} rpm limit={rpm_limit}. current usage={current_rpm}. id={model_id}, model_group={model_group}",
-                            headers={"retry-after": str(60)},
-                            request=httpx.Request(
-                                method="model_rate_limit_check",
-                                url="https://github.com/BerriAI/litellm",
-                            ),
+            if tpd_limit is not None:
+                current_tpd: Final = await self._async_get_current_tpm(tpd_key, tpd_limit, parent_otel_span)
+                if current_tpd is not None and current_tpd >= tpd_limit:
+                    raise daily.limit_error("tpd", tpd_limit, current_tpd, dt)
+
+            charged_rpm_key: Final = rpm_key if rpm_limit is not None else None
+            charged_rpd_key: Final = rpd_key if rpd_limit is not None else None
+            current_rpm, current_rpd = await self._async_increment_request_counts(
+                charged_rpm_key, charged_rpd_key, parent_otel_span
+            )
+            if rpm_limit is not None and current_rpm is not None and current_rpm > rpm_limit:
+                await self._async_refund_request_counts(charged_rpm_key, charged_rpd_key, parent_otel_span)
+                raise litellm.RateLimitError(
+                    message=f"Model rate limit exceeded. RPM limit={rpm_limit}, current usage={current_rpm}",
+                    llm_provider="",
+                    model=model_name,
+                    response=httpx.Response(
+                        status_code=429,
+                        content=f"{RouterErrors.user_defined_ratelimit_error.value} rpm limit={rpm_limit}. current usage={current_rpm}. id={model_id}, model_group={model_group}",
+                        headers={"retry-after": str(60)},
+                        request=httpx.Request(
+                            method="model_rate_limit_check",
+                            url="https://github.com/BerriAI/litellm",
                         ),
-                        num_retries=0,  # Don't retry - return 429 immediately
-                    )
+                    ),
+                    num_retries=0,  # Don't retry - return 429 immediately
+                )
+
+            if rpd_limit is not None and current_rpd is not None and current_rpd > rpd_limit:
+                await self._async_refund_request_counts(charged_rpm_key, charged_rpd_key, parent_otel_span)
+                raise daily.limit_error("rpd", rpd_limit, current_rpd, dt)
 
             return deployment
 
@@ -373,17 +507,16 @@ class ModelRateLimitingCheck(CustomLogger):
             if not model or not total_tokens:
                 return
 
-            dt: Final = get_utc_datetime()
+            dt: Final = self.clock()
             current_minute: Final = dt.strftime("%H-%M")
             tpm_key: Final = f"{model_id}:{model}:tpm:{current_minute}"
+            tpd_key: Final = f"{model_id}:{model}:tpd:{dt.strftime(DAY_FORMAT)}"
 
             verbose_router_logger.debug("[TPM TRACKING] Incrementing %s by %s", tpm_key, total_tokens)
 
-            await self.dual_cache.async_increment_cache(
-                key=tpm_key,
-                value=total_tokens,
-                ttl=RoutingArgs.ttl,
-            )
+            tokens: Final = int(total_tokens)
+            await self._async_increment(tpm_key, tokens, RoutingArgs.ttl)
+            await self._async_increment(tpd_key, tokens, RoutingArgs.daily_ttl)
 
         except Exception as e:
             verbose_router_logger.debug("Error in ModelRateLimitingCheck.async_log_success_event: %s", e)
@@ -434,15 +567,14 @@ class ModelRateLimitingCheck(CustomLogger):
             if not model or not total_tokens:
                 return
 
-            dt: Final = get_utc_datetime()
+            dt: Final = self.clock()
             current_minute: Final = dt.strftime("%H-%M")
             tpm_key: Final = f"{model_id}:{model}:tpm:{current_minute}"
+            tpd_key: Final = f"{model_id}:{model}:tpd:{dt.strftime(DAY_FORMAT)}"
 
-            self.dual_cache.increment_cache(
-                key=tpm_key,
-                value=total_tokens,
-                ttl=RoutingArgs.ttl,
-            )
+            tokens: Final = int(total_tokens)
+            self._increment(tpm_key, tokens, RoutingArgs.ttl)
+            self._increment(tpd_key, tokens, RoutingArgs.daily_ttl)
 
         except Exception as e:
             verbose_router_logger.debug("Error in ModelRateLimitingCheck.log_success_event: %s", e)
diff --git a/litellm/types/litellm_params.py b/litellm/types/litellm_params.py
index 717b74e5741..9196b7a226f 100644
--- a/litellm/types/litellm_params.py
+++ b/litellm/types/litellm_params.py
@@ -189,6 +189,8 @@ class DeploymentOptions:
     tpm: int | None = None
     itpm: int | None = None
     otpm: int | None = None
+    rpd: int | None = None
+    tpd: int | None = None
     kubernetes_pod_discovery: bool | None = None
     default_api_key_rpm_limit: int | None = None
     default_api_key_tpm_limit: int | None = None
diff --git a/litellm/types/router.py b/litellm/types/router.py
index ba9b82e2a57..f8de2128a80 100644
--- a/litellm/types/router.py
+++ b/litellm/types/router.py
@@ -533,6 +533,8 @@ class GenericLiteLLMParams(CredentialLiteLLMParams, CustomPricingLiteLLMParams):
     rpm: int | None = None
     itpm: int | None = None
     otpm: int | None = None
+    tpd: int | None = None
+    rpd: int | None = None
     timeout: float | str | httpx.Timeout | SDKTimeout | None = None  # if str, pass in as os.environ/
     # read timeout of a live stream, a faked stream takes the longer of this and timeout. If str, pass in as os.environ/
     stream_timeout: float | str | None = None
@@ -738,6 +740,8 @@ class LiteLLMParamsTypedDict(TypedDict, total=False):
     rpm: int | None
     itpm: int | None
     otpm: int | None
+    tpd: ReadOnly[int | None]
+    rpd: ReadOnly[int | None]
     order: int | None
     weight: int | None
     max_parallel_requests: int | None
diff --git a/tests/integration/routing/test_enforce_model_rate_limits_daily_limits.py b/tests/integration/routing/test_enforce_model_rate_limits_daily_limits.py
new file mode 100644
index 00000000000..c670171a149
--- /dev/null
+++ b/tests/integration/routing/test_enforce_model_rate_limits_daily_limits.py
@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import uuid
+from collections.abc import Iterator
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final
+
+import httpx
+import pytest
+from integration._support.client import gateway_from_environment
+from integration._support.openai_wire import chat_reply
+from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
+from integration._support.wire import Reply, Request, Wire, wire_server
+from integration.providers._cache_control_marks_support import owned_config
+from integration.routing.test_payment_required_mapping import OPENAI_MODEL, is_model_info_probe, model_list_reply
+from pydantic import JsonValue
+
+pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
+
+GROUP: Final = "daily-limited-group"
+LIMITED_ID: Final = "free-tier-two-a-day"
+UNLIMITED_ID: Final = "paid-tier"
+RPD: Final = 2
+EXTRA_REQUESTS: Final = 2
+
+
+@dataclass(frozen=True, slots=True)
+class Rig:
+    owned: OwnedProxy
+    limited: Wire
+    unlimited: Wire
+
+
+def _reply(request: Request) -> Reply:
+    if is_model_info_probe(request):
+        return model_list_reply()
+    return chat_reply("chatcmpl-daily", OPENAI_MODEL, "served", stream=False)
+
+
+def _deployment(model_id: str, wire: Wire, **params: JsonValue) -> dict[str, JsonValue]:
+    return {
+        "model_name": GROUP,
+        "litellm_params": {
+            "model": OPENAI_MODEL,
+            "api_base": wire.url + "/v1",
+            "api_key": "synthetic-openai-key",
+            **params,
+        },
+        "model_info": {"id": model_id},
+    }
+
+
+def _config(directory: Path, limited: Wire, unlimited: Wire) -> Path:
+    return owned_config(
+        directory,
+        [_deployment(LIMITED_ID, limited, rpd=RPD, order=1), _deployment(UNLIMITED_ID, unlimited, order=2)],
+        litellm_settings={"cache": False},
+        router_settings={"optional_pre_call_checks": ["enforce_model_rate_limits"], "num_retries": 1},
+    )
+
+
+@pytest.fixture(scope="module")
+def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
+    directory: Final = tmp_path_factory.mktemp("daily-limits")
+    with (
+        gateway_from_environment() as gateway,
+        wire_server(_reply) as limited,
+        wire_server(_reply) as unlimited,
+        owned_proxy_process(gateway, directory, {}, config=_config(directory, limited, unlimited)) as owned,
+    ):
+        yield Rig(owned, limited, unlimited)
+
+
+def _chat(rig: Rig) -> httpx.Response:
+    return rig.owned.gateway.request(
+        "POST",
+        "/v1/chat/completions",
+        {"model": GROUP, "messages": [{"role": "user", "content": f"daily {uuid.uuid4().hex}"}]},
+    )
+
+
+def _forwarded(wire: Wire) -> int:
+    return sum(1 for request in wire.drain() if not is_model_info_probe(request))
+
+
+def test_requests_past_the_deployment_rpd_are_served_by_the_other_deployment_in_the_group(rig: Rig) -> None:
+    responses: Final = tuple(_chat(rig) for _ in range(RPD + EXTRA_REQUESTS))
+
+    assert [response.status_code for response in responses] == [200] * (RPD + EXTRA_REQUESTS), [
+        response.text for response in responses
+    ]
+    assert [response.headers["x-litellm-model-id"] for response in responses] == [LIMITED_ID] * RPD + [
+        UNLIMITED_ID
+    ] * EXTRA_REQUESTS
+    assert (_forwarded(rig.limited), _forwarded(rig.unlimited)) == (RPD, EXTRA_REQUESTS)
diff --git a/tests/unit/caching/test_in_memory_cache.py b/tests/unit/caching/test_in_memory_cache.py
index 75232476ecd..901832937b5 100644
--- a/tests/unit/caching/test_in_memory_cache.py
+++ b/tests/unit/caching/test_in_memory_cache.py
@@ -235,13 +235,9 @@ def test_in_memory_cache_eviction_order():
 
     # Add items with different TTLs
     now = time.time()
-    in_memory_cache.set_cache(
-        key="early_expire", value="value_1", ttl=100
-    )  # expires in 100 seconds
+    in_memory_cache.set_cache(key="early_expire", value="value_1", ttl=100)  # expires in 100 seconds
     time.sleep(0.01)
-    in_memory_cache.set_cache(
-        key="late_expire", value="value_2", ttl=200
-    )  # expires in 200 seconds
+    in_memory_cache.set_cache(key="late_expire", value="value_2", ttl=200)  # expires in 200 seconds
 
     # Verify TTL order
     early_ttl = in_memory_cache.ttl_dict["early_expire"]
@@ -379,3 +375,20 @@ def test_in_memory_cache_injected_clock_controls_expiry_and_eviction() -> None:
     assert cache.get_cache("second") is None
     assert cache.get_cache("third") == "replacement"
     assert cache.get_cache("fourth") == "new"
+
+
+@pytest.mark.asyncio
+async def test_async_increment_pipeline_keeps_each_key_for_its_own_ttl() -> None:
+    now: Final = {"t": 1_000.0}
+    cache: Final = InMemoryCache(default_ttl=600, clock=lambda: now["t"])
+
+    await cache.async_increment_pipeline(
+        increment_list=[
+            {"key": "minute", "increment_value": 1, "ttl": 60},
+            {"key": "day", "increment_value": 1, "ttl": 86400},
+        ]
+    )
+    now["t"] += 601
+
+    assert cache.get_cache("minute") is None
+    assert cache.get_cache("day") == 1
diff --git a/tests/unit/test_router/test_enforce_model_rate_limits.py b/tests/unit/test_router/test_enforce_model_rate_limits.py
index 7577064b7f9..2a260fdf0b8 100644
--- a/tests/unit/test_router/test_enforce_model_rate_limits.py
+++ b/tests/unit/test_router/test_enforce_model_rate_limits.py
@@ -6,7 +6,7 @@ regardless of the routing strategy being used.
 """
 
 import asyncio
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -222,11 +222,9 @@ class TestModelRateLimitingCheck:
 
         check.log_success_event(kwargs, None, None, None)
 
-        # Verify increment_cache was called
-        mock_cache.increment_cache.assert_called_once()
-        _, kwarg_params = mock_cache.increment_cache.call_args
-        assert "test-id:gpt-4:tpm:" in kwarg_params["key"]
-        assert kwarg_params["value"] == 50
+        keys = tuple(call.kwargs["key"] for call in mock_cache.increment_cache.call_args_list)
+        assert any(key.startswith("test-id:gpt-4:tpm:") for key in keys), keys
+        assert all(call.kwargs["value"] == 50 for call in mock_cache.increment_cache.call_args_list)
 
 
 class TestModelRateLimitingCheckAsync:
@@ -253,7 +251,7 @@ class TestModelRateLimitingCheckAsync:
         """Test that RateLimitError is raised when RPM limit is exceeded (async)."""
         mock_cache = MagicMock()
         mock_cache.async_get_cache = AsyncMock(return_value=None)
-        mock_cache.async_increment_cache = AsyncMock(return_value=11)  # Over limit
+        mock_cache.async_increment_cache_pipeline = AsyncMock(return_value=[11.0])  # Over limit
 
         check = ModelRateLimitingCheck(dual_cache=mock_cache)
 
@@ -274,7 +272,7 @@ class TestModelRateLimitingCheckAsync:
         """Test that requests are allowed when under the limit (async)."""
         mock_cache = MagicMock()
         mock_cache.async_get_cache = AsyncMock(return_value=None)
-        mock_cache.async_increment_cache = AsyncMock(return_value=6)
+        mock_cache.async_increment_cache_pipeline = AsyncMock(return_value=[6.0])
 
         check = ModelRateLimitingCheck(dual_cache=mock_cache)
 
@@ -339,7 +337,7 @@ class TestModelRateLimitingCheckAsync:
     ):
         redis_cache = MagicMock()
         redis_cache.async_get_cache = AsyncMock(side_effect=RedisCircuitBreakerOpenError())
-        redis_cache.async_increment = AsyncMock(return_value=2)
+        redis_cache.async_increment_pipeline = AsyncMock(return_value=[2.0])
         check = ModelRateLimitingCheck(dual_cache=_dual_cache_with_local_tpm(5, redis_cache))
 
         with pytest.raises(litellm.RateLimitError) as exc_info:
@@ -375,11 +373,9 @@ class TestModelRateLimitingCheckAsync:
 
         await check.async_log_success_event(kwargs, None, None, None)
 
-        # Verify async_increment_cache was called
-        mock_cache.async_increment_cache.assert_called_once()
-        _, kwarg_params = mock_cache.async_increment_cache.call_args
-        assert "test-id:gpt-4:tpm:" in kwarg_params["key"]
-        assert kwarg_params["value"] == 50
+        keys = tuple(call.kwargs["key"] for call in mock_cache.async_increment_cache.call_args_list)
+        assert any(key.startswith("test-id:gpt-4:tpm:") for key in keys), keys
+        assert all(call.kwargs["value"] == 50 for call in mock_cache.async_increment_cache.call_args_list)
 
 
 class TestRouterWithEnforceModelRateLimits:
@@ -466,3 +462,265 @@ class TestModelRateLimitConcurrency:
 
         assert len(successes) == 2, f"Expected 2 successes, got {len(successes)}"
         assert len(failures) == 2, f"Expected 2 rate limit errors, got {len(failures)}"
+
+
+DAY_ONE = datetime(2026, 10, 28, 23, 59, 30, tzinfo=timezone.utc)
+RPD_DEPLOYMENT = {
+    "litellm_params": {"model": "gpt-4.1-mini", "rpd": 2},
+    "model_info": {"id": "daily-test-id"},
+    "model_name": "daily-model",
+}
+RPM_AND_RPD_DEPLOYMENT = {**RPD_DEPLOYMENT, "litellm_params": {"model": "gpt-4.1-mini", "rpm": 1, "rpd": 2}}
+
+
+class Clock:
+    def __init__(self, now: datetime):
+        self.now = now
+
+    def __call__(self) -> datetime:
+        return self.now
+
+
+def _pre_call(check: ModelRateLimitingCheck, deployment: dict[str, object]) -> object:
+    return check.pre_call_check(deployment)
+
+
+async def _async_pre_call(check: ModelRateLimitingCheck, deployment: dict[str, object]) -> object:
+    return await check.async_pre_call_check(deployment)
+
+
+async def _async_log_success(check: ModelRateLimitingCheck, kwargs: dict[str, object]) -> None:
+    await check.async_log_success_event(kwargs, None, None, None)
+
+
+class TestDailyLimits:
+    def test_pre_call_check_rejects_the_request_over_rpd(self):
+        check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=Clock(DAY_ONE))
+
+        assert _pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
+        assert _pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
+        with pytest.raises(litellm.RateLimitError) as exc_info:
+            _pre_call(check, RPD_DEPLOYMENT)
+
+        assert "RPD limit=2, current usage=3" in str(exc_info.value)
+        assert "rpd limit=2. current usage=3. id=daily-test-id, model_group=daily-model" in exc_info.value.response.text
+
+    @pytest.mark.asyncio
+    async def test_async_pre_call_check_rejects_the_request_over_rpd_and_lets_the_router_retry(self):
+        check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=Clock(DAY_ONE))
+
+        assert await _async_pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
+        assert await _async_pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
+        with pytest.raises(litellm.RateLimitError) as exc_info:
+            await _async_pre_call(check, RPD_DEPLOYMENT)
+
+        assert "RPD limit=2" in str(exc_info.value)
+        assert exc_info.value.num_retries is None
+
+    @pytest.mark.asyncio
+    async def test_async_pre_call_check_rejects_once_the_day_tokens_reach_tpd(self):
+        deployment = {**RPD_DEPLOYMENT, "litellm_params": {"model": "gpt-4.1-mini", "tpd": 100}}
+        check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=Clock(DAY_ONE))
+        kwargs = {
+            "standard_logging_object": {
+                "model_id": "daily-test-id",
+                "total_tokens": 60,
+                "hidden_params": {"litellm_model_name": "gpt-4.1-mini"},
+            }
+        }
+
+        await _async_log_success(check, kwargs)
+        assert await _async_pre_call(check, deployment) == deployment
+        await _async_log_success(check, kwargs)
+        with pytest.raises(litellm.RateLimitError) as exc_info:
+            await _async_pre_call(check, deployment)
+
+        assert "TPD limit=100, current usage=120" in str(exc_info.value)
+
+    def test_pre_call_check_minute_rejected_request_does_not_spend_rpd(self):
+        clock = Clock(DAY_ONE)
+        check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
+
+        assert _pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
+        with pytest.raises(litellm.RateLimitError, match="RPM limit=1"):
+            _pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+        clock.now = DAY_ONE - timedelta(minutes=1)
+
+        assert _pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
+        clock.now = DAY_ONE - timedelta(minutes=2)
+        with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
+            _pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+
+    @pytest.mark.asyncio
+    async def test_async_pre_call_check_minute_rejected_request_does_not_spend_rpd(self):
+        clock = Clock(DAY_ONE)
+        check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
+
+        assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
+        with pytest.raises(litellm.RateLimitError, match="RPM limit=1"):
+            await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+        clock.now = DAY_ONE - timedelta(minutes=1)
+
+        assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
+        clock.now = DAY_ONE - timedelta(minutes=2)
+        with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
+            await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+
+    def test_pre_call_check_rejected_requests_hand_back_rpm_and_rpd(self):
+        clock = Clock(DAY_ONE)
+        check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
+
+        assert _pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
+        with pytest.raises(litellm.RateLimitError, match="RPM limit=1, current usage=2"):
+            _pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+        with pytest.raises(litellm.RateLimitError, match="RPM limit=1, current usage=2"):
+            _pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+        clock.now = DAY_ONE - timedelta(minutes=1)
+        assert _pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
+
+        clock.now = DAY_ONE - timedelta(minutes=2)
+        with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
+            _pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+        with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
+            _pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+
+    @pytest.mark.asyncio
+    async def test_async_pre_call_check_rejected_requests_hand_back_rpm_and_rpd(self):
+        clock = Clock(DAY_ONE)
+        check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
+
+        assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
+        with pytest.raises(litellm.RateLimitError, match="RPM limit=1, current usage=2"):
+            await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+        with pytest.raises(litellm.RateLimitError, match="RPM limit=1, current usage=2"):
+            await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+        clock.now = DAY_ONE - timedelta(minutes=1)
+        assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
+
+        clock.now = DAY_ONE - timedelta(minutes=2)
+        with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
+            await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+        with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
+            await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+
+    @pytest.mark.asyncio
+    async def test_async_pre_call_check_refunds_rpm_and_rpd_in_one_cache_trip(self):
+        mock_cache = MagicMock()
+        mock_cache.async_get_cache = AsyncMock(return_value=None)
+        mock_cache.async_increment_cache_pipeline = AsyncMock(side_effect=[[1.0, 3.0], [0.0, 2.0]])
+        check = ModelRateLimitingCheck(dual_cache=mock_cache, clock=Clock(DAY_ONE))
+
+        with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
+            await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
+
+        assert mock_cache.async_increment_cache_pipeline.await_count == 2
+        refund = mock_cache.async_increment_cache_pipeline.call_args_list[1].kwargs["increment_list"]
+        assert [(op["key"], op["increment_value"], op["ttl"]) for op in refund] == [
+            ("daily-test-id:gpt-4.1-mini:rpm:23-59", -1, 60),
+            ("daily-test-id:gpt-4.1-mini:rpd:2026-10-28", -1, 86400),
+        ]
+
+    @pytest.mark.asyncio
+    async def test_async_pre_call_check_increments_rpm_and_rpd_in_one_cache_trip(self):
+        mock_cache = MagicMock()
+        mock_cache.async_get_cache = AsyncMock(return_value=None)
+        mock_cache.async_increment_cache_pipeline = AsyncMock(return_value=[1.0, 1.0])
+        check = ModelRateLimitingCheck(dual_cache=mock_cache, clock=Clock(DAY_ONE))
+
+        assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
+
+        assert mock_cache.async_increment_cache_pipeline.await_count == 1
+        operations = mock_cache.async_increment_cache_pipeline.call_args.kwargs["increment_list"]
+        assert [(op["key"], op["ttl"]) for op in operations] == [
+            ("daily-test-id:gpt-4.1-mini:rpm:23-59", 60),
+            ("daily-test-id:gpt-4.1-mini:rpd:2026-10-28", 86400),
+        ]
+        assert not mock_cache.async_increment_cache.called
+
+    def test_pre_call_check_reads_the_shared_tpd_count_from_redis(self):
+        redis_cache = MagicMock()
+        redis_cache.get_cache.return_value = 100
+        deployment = {**RPD_DEPLOYMENT, "litellm_params": {"model": "gpt-4.1-mini", "tpd": 100}}
+        check = ModelRateLimitingCheck(dual_cache=DualCache(redis_cache=redis_cache), clock=Clock(DAY_ONE))
+
+        with pytest.raises(litellm.RateLimitError) as exc_info:
+            _pre_call(check, deployment)
+
+        assert "TPD limit=100, current usage=100" in str(exc_info.value)
+        assert redis_cache.get_cache.call_args.kwargs["key"] == "daily-test-id:gpt-4.1-mini:tpd:2026-10-28"
+
+    def test_rpd_count_starts_fresh_after_utc_midnight(self):
+        clock = Clock(DAY_ONE)
+        check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
+        _pre_call(check, RPD_DEPLOYMENT)
+        _pre_call(check, RPD_DEPLOYMENT)
+        with pytest.raises(litellm.RateLimitError):
+            _pre_call(check, RPD_DEPLOYMENT)
+
+        clock.now = DAY_ONE + timedelta(seconds=30)
+
+        assert _pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
+        assert _pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
+        with pytest.raises(litellm.RateLimitError) as exc_info:
+            _pre_call(check, RPD_DEPLOYMENT)
+        assert "current usage=3" in str(exc_info.value)
+
+    @pytest.mark.parametrize(
+        ("now", "seconds_to_midnight"),
+        [
+            (DAY_ONE, 30),
+            (datetime(2026, 10, 28, 0, 0, 0, tzinfo=timezone.utc), 86400),
+            (datetime(2026, 10, 28, 12, 30, 15, 500000, tzinfo=timezone.utc), 41384),
+        ],
+    )
+    def test_daily_429_retry_after_is_seconds_until_utc_midnight(self, now: datetime, seconds_to_midnight: int):
+        deployment = {**RPD_DEPLOYMENT, "litellm_params": {"model": "gpt-4.1-mini", "rpd": 0}}
+        check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=Clock(now))
+
+        with pytest.raises(litellm.RateLimitError) as exc_info:
+            _pre_call(check, deployment)
+
+        assert exc_info.value.response.headers["retry-after"] == str(seconds_to_midnight)
+
+    @pytest.mark.asyncio
+    async def test_router_serves_the_request_from_the_other_deployment_once_rpd_is_spent(self):
+        router = Router(
+            model_list=[
+                {
+                    "model_name": "daily-group",
+                    "litellm_params": {
+                        "model": "openai/gpt-4.1-mini",
+                        "api_key": "k",
+                        "mock_response": "a",
+                        "rpd": 1,
+                        "order": 1,
+                    },
+                    "model_info": {"id": "spent-first"},
+                },
+                {
+                    "model_name": "daily-group",
+                    "litellm_params": {
+                        "model": "openai/gpt-4.1-mini",
+                        "api_key": "k",
+                        "mock_response": "b",
+                        "order": 2,
+                    },
+                    "model_info": {"id": "always-free"},
+                },
+            ],
+            num_retries=1,
+        )
+        check = ModelRateLimitingCheck(dual_cache=router.cache, clock=Clock(DAY_ONE))
+        router.optional_callbacks = [check]
+        litellm.logging_callback_manager.add_litellm_callback(check)
+        try:
+            served = [
+                (await router.acompletion(model="daily-group", messages=[{"role": "user", "content": "hi"}]))
+                .choices[0]
+                .message.content
+                for _ in range(4)
+            ]
+        finally:
+            router.discard()
+
+        assert served == ["a", "b", "b", "b"]
diff --git a/tests/unit/test_router/test_io_token_rate_limits.py b/tests/unit/test_router/test_io_token_rate_limits.py
index a5a68271111..0ae22534278 100644
--- a/tests/unit/test_router/test_io_token_rate_limits.py
+++ b/tests/unit/test_router/test_io_token_rate_limits.py
@@ -859,7 +859,7 @@ class TestModelRateLimitingCheckIOTokens:
             with pytest.raises(litellm.RateLimitError):
                 await check.async_pre_call_check(deployment)
 
-        assert await dual_cache.async_get_cache(key=rpm_key) == 6
+        assert await dual_cache.async_get_cache(key=rpm_key) == 5, "the rejected request hands its rpm count back"
         assert (await dual_cache.async_get_cache(key=itpm_key) or 0) == 0
         assert ITPM_RESERVED_KEY not in request_kwargs["metadata"]
         assert any("both limit types are enforced" in record.message for record in caplog.records)
diff --git a/tests/unit/types/test_litellm_params.py b/tests/unit/types/test_litellm_params.py
index 740342f54d9..cc1ea645012 100644
--- a/tests/unit/types/test_litellm_params.py
+++ b/tests/unit/types/test_litellm_params.py
@@ -154,6 +154,8 @@ OPTION_NAMES: Final = (
     "model_info",
     "rpm",
     "tpm",
+    "rpd",
+    "tpd",
     "itpm",
     "otpm",
     "default_api_key_rpm_limit",
diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts
index a09672d7d8f..a49f1099121 100644
--- a/ui/litellm-dashboard/src/lib/http/schema.d.ts
+++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts
@@ -35647,6 +35647,8 @@ export interface components {
             regional_processing_uplift_multiplier_us?: number | null;
             /** Responses Api Version */
             responses_api_version?: string | null;
+            /** Rpd */
+            rpd?: number | null;
             /** Rpm */
             rpm?: number | null;
             /** S3 Access Key Id */
@@ -35699,6 +35701,8 @@ export interface components {
             token_exchange_profile?: string | null;
             /** Token Exchange Scope */
             token_exchange_scope?: string | null;
+            /** Tpd */
+            tpd?: number | null;
             /** Tpm */
             tpm?: number | null;
             /** Use Chat Completions Api */
@@ -51039,6 +51043,8 @@ export interface components {
             regional_processing_uplift_multiplier_us?: number | null;
             /** Responses Api Version */
             responses_api_version?: string | null;
+            /** Rpd */
+            rpd?: number | null;
             /** Rpm */
             rpm?: number | null;
             /** S3 Access Key Id */
@@ -51091,6 +51097,8 @@ export interface components {
             token_exchange_profile?: string | null;
             /** Token Exchange Scope */
             token_exchange_scope?: string | null;
+            /** Tpd */
+            tpd?: number | null;
             /** Tpm */
             tpm?: number | null;
             /** Use Chat Completions Api */