diff --git a/backend/app/api/pricing.py b/backend/app/api/pricing.py
index b1bd65f..78d6386 100644
--- a/backend/app/api/pricing.py
+++ b/backend/app/api/pricing.py
@@ -15,6 +15,7 @@ from fastapi import APIRouter, Depends, Query
 from sqlalchemy.orm import Session
 
 from backend.app.api.deps import get_current_user, require_permissions, require_roles
+from backend.app.core.cache import cache_delete_pattern, cache_get, cache_set
 from backend.app.core.exceptions import AppException
 from backend.app.core.error_codes import ErrorCode
 from backend.app.db import get_db_session
@@ -88,13 +89,19 @@ def get_pricing_rule(
     返回值：定价规则详情信息。
     权限要求：管理员（admin）或经理（manager）。
     """
+    cache_key = f"pricing:rule:{product_id}"
+    cached = cache_get(cache_key)
+    if cached is not None:
+        return success_payload(cached)
     rule = session.query(ProductPricingRule).filter(
         ProductPricingRule.product_id == product_id,
         ProductPricingRule.deleted == 0,
     ).first()
     if not rule:
         raise AppException(code=ErrorCode.NOT_FOUND, message="定价规则不存在", status_code=404)
-    return success_payload(_rule_to_dict(rule))
+    rule_data = _rule_to_dict(rule)
+    cache_set(cache_key, rule_data, ttl=300)
+    return success_payload(rule_data)
 
 
 @router.post("/pricing-rules")
@@ -132,6 +139,7 @@ def create_pricing_rule(
     )
     session.add(rule)
     session.commit()
+    cache_delete_pattern("pricing:*")
     session.refresh(rule)
     return success_payload(_rule_to_dict(rule))
 
@@ -161,6 +169,7 @@ def update_pricing_rule(
     for k, v in data.items():
         setattr(rule, k, v)
     session.commit()
+    cache_delete_pattern("pricing:*")
     session.refresh(rule)
     return success_payload(_rule_to_dict(rule))
 
@@ -187,6 +196,7 @@ def delete_pricing_rule(
         raise AppException(code=ErrorCode.NOT_FOUND, message="定价规则不存在", status_code=404)
     rule.deleted = 1
     session.commit()
+    cache_delete_pattern("pricing:*")
     return success_payload({"deleted": True})
 
 
@@ -230,6 +240,7 @@ def create_supplier_cost(
     cost = SupplierProductCost(**payload.model_dump())
     session.add(cost)
     session.commit()
+    cache_delete_pattern("pricing:*")
     session.refresh(cost)
     return success_payload(_cost_to_dict(cost))
 
@@ -258,6 +269,7 @@ def update_supplier_cost(
     for k, v in data.items():
         setattr(cost, k, v)
     session.commit()
+    cache_delete_pattern("pricing:*")
     session.refresh(cost)
     return success_payload(_cost_to_dict(cost))
 
@@ -283,6 +295,7 @@ def delete_supplier_cost(
         raise AppException(code=ErrorCode.NOT_FOUND, message="供应商成本不存在", status_code=404)
     cost.deleted = 1
     session.commit()
+    cache_delete_pattern("pricing:*")
     return success_payload({"deleted": True})
 
 
@@ -326,6 +339,7 @@ def create_price_tier(
     tier = ProductPriceTier(**payload.model_dump())
     session.add(tier)
     session.commit()
+    cache_delete_pattern("pricing:*")
     session.refresh(tier)
     return success_payload(_tier_to_dict(tier))
 
@@ -354,6 +368,7 @@ def update_price_tier(
     for k, v in data.items():
         setattr(tier, k, v)
     session.commit()
+    cache_delete_pattern("pricing:*")
     session.refresh(tier)
     return success_payload(_tier_to_dict(tier))
 
@@ -379,6 +394,7 @@ def delete_price_tier(
         raise AppException(code=ErrorCode.NOT_FOUND, message="价格层级不存在", status_code=404)
     tier.deleted = 1
     session.commit()
+    cache_delete_pattern("pricing:*")
     return success_payload({"deleted": True})
 
 
diff --git a/backend/app/core/cache.py b/backend/app/core/cache.py
new file mode 100644
index 0000000..4fcb33a
--- /dev/null
+++ b/backend/app/core/cache.py
@@ -0,0 +1,105 @@
+"""Redis 缓存工具模块。
+
+提供 Redis 连接管理和缓存读写操作，所有操作在 Redis 不可用时静默降级。
+被 OrderService、CustomerService、pricing 路由等调用。
+"""
+
+import hashlib
+import json
+import logging
+
+import redis
+
+from backend.app.core.config import get_settings
+
+logger = logging.getLogger(__name__)
+
+_redis_client: redis.Redis | None = None
+
+
+def get_redis() -> redis.Redis | None:
+    """获取 Redis 连接单例，连接失败时返回 None。"""
+    global _redis_client
+    if _redis_client is not None:
+        return _redis_client
+    try:
+        settings = get_settings()
+        _redis_client = redis.Redis(
+            host=settings.redis_host,
+            port=settings.redis_port,
+            db=settings.redis_db,
+            password=settings.redis_password or None,
+            decode_responses=True,
+            socket_connect_timeout=2,
+            socket_timeout=2,
+        )
+        _redis_client.ping()
+        logger.info("Redis 连接成功: %s:%s/%s", settings.redis_host, settings.redis_port, settings.redis_db)
+        return _redis_client
+    except Exception as exc:
+        logger.warning("Redis 连接失败，降级为无缓存模式: %s", exc)
+        _redis_client = None
+        return None
+
+
+def cache_get(key: str):
+    """读取缓存，返回反序列化后的 dict/list，未命中或异常返回 None。"""
+    try:
+        r = get_redis()
+        if r is None:
+            return None
+        raw = r.get(key)
+        if raw is None:
+            return None
+        return json.loads(raw)
+    except Exception:
+        return None
+
+
+def cache_set(key: str, data, ttl: int = 60) -> None:
+    """写入缓存，JSON 序列化后存储并设置 TTL（秒）。异常时静默跳过。"""
+    try:
+        r = get_redis()
+        if r is None:
+            return
+        r.set(key, json.dumps(data, ensure_ascii=False, default=str), ex=ttl)
+    except Exception:
+        pass
+
+
+def cache_delete(key: str) -> None:
+    """精确删除单个缓存 key。"""
+    try:
+        r = get_redis()
+        if r is not None:
+            r.delete(key)
+    except Exception:
+        pass
+
+
+def cache_delete_pattern(pattern: str) -> None:
+    """按模式批量删除缓存（SCAN + DELETE），如 "order:list:*"。"""
+    try:
+        r = get_redis()
+        if r is None:
+            return
+        cursor = 0
+        while True:
+            cursor, keys = r.scan(cursor=cursor, match=pattern, count=100)
+            if keys:
+                r.delete(*keys)
+            if cursor == 0:
+                break
+    except Exception:
+        pass
+
+
+def make_cache_key(prefix: str, **params) -> str:
+    """生成带参数哈希的缓存 key。
+
+    例: make_cache_key("order:list", user_id=1, status="draft")
+    → "order:list:a1b2c3d4"
+    """
+    raw = json.dumps(params, sort_keys=True, ensure_ascii=False, default=str)
+    suffix = hashlib.md5(raw.encode()).hexdigest()[:8]
+    return f"{prefix}:{suffix}"
diff --git a/backend/app/core/config.py b/backend/app/core/config.py
index 5ae9695..af0233c 100644
--- a/backend/app/core/config.py
+++ b/backend/app/core/config.py
@@ -47,6 +47,10 @@ class Settings(BaseSettings):
     mysql_user: str = Field(default="order_user", alias="MYSQL_USER")
     mysql_password: str = Field(default="password", alias="MYSQL_PASSWORD")
     sql_echo: bool = Field(default=False, alias="SQL_ECHO")
+    redis_host: str = Field(default="127.0.0.1", alias="REDIS_HOST")
+    redis_port: int = Field(default=6379, alias="REDIS_PORT")
+    redis_db: int = Field(default=0, alias="REDIS_DB")
+    redis_password: str = Field(default="", alias="REDIS_PASSWORD")
     aliyun_oss_bucket: str = Field(default="bucket-name", alias="ALIYUN_OSS_BUCKET")
     aliyun_oss_endpoint: str = Field(default="oss-cn-hangzhou.aliyuncs.com", alias="ALIYUN_OSS_ENDPOINT")
     aliyun_oss_public_base_url: str = Field(default="https://cdn.example.com", alias="ALIYUN_OSS_PUBLIC_BASE_URL")
diff --git a/backend/app/services/customer_service.py b/backend/app/services/customer_service.py
index 722e1e2..f7c36af 100644
--- a/backend/app/services/customer_service.py
+++ b/backend/app/services/customer_service.py
@@ -11,6 +11,7 @@ from pathlib import Path
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session
 
+from backend.app.core.cache import cache_delete_pattern, cache_get, cache_set, make_cache_key
 from backend.app.core.error_codes import ErrorCode
 from backend.app.core.exceptions import AppException
 from backend.app.repositories.customer_repository import CustomerRepository
@@ -44,9 +45,14 @@ class CustomerService:
         """
         if session is None:
             raise AppException(code=ErrorCode.PARAM_ERROR, message="数据库会话未初始化", status_code=500)
+        # 读缓存
+        cache_key = make_cache_key("customer:list", **(filters or {}))
+        cached = cache_get(cache_key)
+        if cached is not None:
+            return cached
         try:
             customers = self.repository.list_customers_by_filters(session, filters or {})
-            return {
+            result = {
                 "total": len(customers),
                 "page_no": 1,
                 "page_size": 20,
@@ -67,6 +73,8 @@ class CustomerService:
                     for item in customers
                 ],
             }
+            cache_set(cache_key, result, ttl=60)
+            return result
         except SQLAlchemyError as exc:
             raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库操作失败", status_code=500) from exc
 
@@ -136,6 +144,7 @@ class CustomerService:
                 },
             )
             session.commit()
+            cache_delete_pattern("customer:list:*")
             return {
                 "customer_id": customer.id,
                 "customer_name": customer.customer_name,
@@ -212,6 +221,7 @@ class CustomerService:
                 },
             )
             session.commit()
+            cache_delete_pattern("customer:list:*")
             return {"customer_id": customer.id, "customer_name": customer.customer_name, "mobile": customer.mobile}
         except AppException:
             session.rollback()
@@ -314,6 +324,7 @@ class CustomerService:
                 },
             )
             session.commit()
+            cache_delete_pattern("customer:list:*")
             return result
         except AppException:
             session.rollback()
diff --git a/backend/app/services/order_service.py b/backend/app/services/order_service.py
index f246c06..154fce9 100644
--- a/backend/app/services/order_service.py
+++ b/backend/app/services/order_service.py
@@ -12,6 +12,7 @@ import json
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session
 
+from backend.app.core.cache import cache_delete, cache_delete_pattern, cache_get, cache_set, make_cache_key
 from backend.app.core.error_codes import ErrorCode
 from backend.app.core.exceptions import AppException
 from backend.app.repositories.customer_repository import CustomerRepository
@@ -65,11 +66,16 @@ class OrderService:
         被调用路由: orders.py - GET /orders, salesman.py - GET /salesman/orders
         """
         if session is not None:
+            # 读缓存
+            cache_key = make_cache_key("order:list", user_id=(current_user or {}).get("user_id"), **(filters or {}))
+            cached = cache_get(cache_key)
+            if cached is not None:
+                return cached
             try:
                 result = self.order_repository.list_orders_by_filters(session, filters or {})
                 orders = result["list"]
                 role_code = (current_user or {}).get("role_code")
-                return [
+                rows = [
                     self._filter_order_row(
                         {
                             "order_id": item.id,
@@ -88,6 +94,9 @@ class OrderService:
                     )
                     for item in orders
                 ]
+                # 写缓存
+                cache_set(cache_key, rows, ttl=30)
+                return rows
             except SQLAlchemyError as exc:
                 raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库异常", status_code=500) from exc
         return []
@@ -114,6 +123,11 @@ class OrderService:
         被调用路由: orders.py - GET /orders/{order_id}
         """
         if session is not None:
+            # 读缓存
+            cache_key = f"order:detail:{order_id}"
+            cached = cache_get(cache_key)
+            if cached is not None:
+                return self._filter_order_detail_by_role(cached, current_user)
             try:
                 order = self.order_repository.get_order(session, order_id)
                 if order is None:
@@ -213,6 +227,8 @@ class OrderService:
                     "profit_alert_threshold": float(self._get_config_value(session, "profit_alert_threshold", "0")),
                     "remark": order.remark,
                 }
+                # 写缓存（缓存完整结果，角色过滤在读取后执行）
+                cache_set(cache_key, result, ttl=60)
                 return self._filter_order_detail_by_role(result, current_user)
             except SQLAlchemyError as exc:
                 raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库异常", status_code=500) from exc
@@ -325,6 +341,7 @@ class OrderService:
                     },
                 )
                 session.commit()
+                self._invalidate_order_cache(order.id)
                 return {
                     "order_id": order.id,
                     "order_no": order.order_no,
@@ -388,6 +405,7 @@ class OrderService:
                 )
                 self._notify_status_change(session, order, before_status, order.order_status, current_user.get("user_id") if current_user else None)
                 session.commit()
+                self._invalidate_order_cache(order.id)
                 return {
                     "order_id": order.id,
                     "order_status": order.order_status,
@@ -487,6 +505,7 @@ class OrderService:
                 )
                 self._notify_status_change(session, order, previous_status, order.order_status, current_user.get("user_id") if current_user else None)
                 session.commit()
+                self._invalidate_order_cache(order.id)
                 return {
                     "order_id": order.id,
                     "order_status": order.order_status,
@@ -567,6 +586,7 @@ class OrderService:
                 )
                 self._notify_status_change(session, order, previous_status, order.order_status, current_user.get("user_id") if current_user else None, {"approve_opinion": payload.get("approve_opinion")})
                 session.commit()
+                self._invalidate_order_cache(order.id)
                 return {
                     "order_id": order.id,
                     "order_status": order.order_status,
@@ -655,6 +675,7 @@ class OrderService:
                 )
                 self._notify_status_change(session, order, previous_status, order.order_status, current_user.get("user_id") if current_user else None)
                 session.commit()
+                self._invalidate_order_cache(order.id)
                 return {
                     "order_id": order.id,
                     "order_status": order.order_status,
@@ -782,6 +803,7 @@ class OrderService:
                     },
                 )
                 session.commit()
+                self._invalidate_order_cache(order.id)
                 return {
                     "order_id": order.id,
                     "order_status": order.order_status,
@@ -864,6 +886,7 @@ class OrderService:
                 )
                 self._notify_status_change(session, order, before_status, order.order_status, current_user.get("user_id") if current_user else None)
                 session.commit()
+                self._invalidate_order_cache(order.id)
                 return {
                     "order_id": order.id,
                     "order_status": order.order_status,
@@ -988,6 +1011,7 @@ class OrderService:
                     },
                 )
                 session.commit()
+                self._invalidate_order_cache(order.id)
                 return {
                     "order_id": order.id,
                     "order_no": order.order_no,
@@ -1199,6 +1223,12 @@ class OrderService:
             return filtered
         return order
 
+    def _invalidate_order_cache(self, order_id: int | None = None) -> None:
+        """清除订单相关缓存。"""
+        if order_id:
+            cache_delete(f"order:detail:{order_id}")
+        cache_delete_pattern("order:list:*")
+
     def _filter_order_row(self, row: dict, role_code: str | None) -> dict:
         """根据用户角色过滤订单列表行中的敏感字段。