diff --git a/app/api/routers/rbac_opc.py b/app/api/routers/rbac_opc.py
index 8f51a80..da9c8f4 100644
--- a/app/api/routers/rbac_opc.py
+++ b/app/api/routers/rbac_opc.py
@@ -584,6 +584,77 @@ async def opc_compute_subsidy_records(
     return {"items": items, "total": total, "p": p, "page_size": page_size}
 
 
+@router.get("/compute/subsidy/eligibility/status", summary="补贴资格认证状态")
+async def opc_compute_subsidy_eligibility_status(
+    user: dict = Depends(require_roles("opc_member", "operator")),
+):
+    """查询当前用户的补贴资格认证状态。"""
+    try:
+        data = await compute_client.user_eligibility_status(user.get("username"))
+    except compute_client.ComputeError as exc:
+        raise HTTPException(status_code=502, detail=f"算力引擎对接失败: {exc}") from exc
+    return data
+
+
+class EligibilityApplyBody(BaseModel):
+    cert_type: str
+    real_name: str
+    id_card: str = ""
+    phone: str = ""
+    company_name: str = ""
+    company_credit_code: str = ""
+    province: str = "云南省"
+    city: str = ""
+    address: str = ""
+    materials: list = []
+
+
+@router.post("/compute/subsidy/eligibility/apply", summary="提交补贴资格认证")
+async def opc_compute_subsidy_eligibility_apply(
+    body: EligibilityApplyBody,
+    user: dict = Depends(require_roles("opc_member", "operator")),
+):
+    """提交补贴资格认证申请（云南户籍/企业/就业创业三选一）。"""
+    try:
+        data = await compute_client.user_eligibility_apply(user.get("username"), body.model_dump())
+    except compute_client.ComputeError as exc:
+        raise HTTPException(status_code=502, detail=f"算力引擎对接失败: {exc}") from exc
+    return data
+
+
+class SubsidyApplyBody(BaseModel):
+    period: str = ""
+    note: str = ""
+
+
+@router.post("/compute/subsidy/apply", summary="手动申请补贴")
+async def opc_compute_subsidy_apply(
+    body: SubsidyApplyBody,
+    user: dict = Depends(require_roles("opc_member", "operator")),
+):
+    """手动申请周期补贴（需资格认证通过）。"""
+    try:
+        data = await compute_client.user_subsidy_apply(
+            user.get("username"), period=body.period, note=body.note,
+        )
+    except compute_client.ComputeError as exc:
+        raise HTTPException(status_code=502, detail=f"算力引擎对接失败: {exc}") from exc
+    return data
+
+
+@router.get("/compute/subsidy/calc-preview", summary="补贴金额预览")
+async def opc_compute_subsidy_calc_preview(
+    period: str = "",
+    user: dict = Depends(require_roles("opc_member", "operator")),
+):
+    """手动申请前的补贴金额预览（不生成记录）。"""
+    try:
+        data = await compute_client.user_subsidy_calc_preview(user.get("username"), period=period)
+    except compute_client.ComputeError as exc:
+        raise HTTPException(status_code=502, detail=f"算力引擎对接失败: {exc}") from exc
+    return data
+
+
 @router.get("/compute/tokens", summary="我的算力令牌")
 async def opc_compute_tokens(user: dict = Depends(require_roles("opc_member", "operator"))):
     """列出当前用户的引擎消费令牌。key 统一补 sk- 前缀（OpenAI 客户端口径）。"""
diff --git a/app/services/compute_client.py b/app/services/compute_client.py
index ae23733..44fe867 100644
--- a/app/services/compute_client.py
+++ b/app/services/compute_client.py
@@ -222,6 +222,56 @@ async def user_subsidy_records(username: str, p: int = 1, page_size: int = 20) -
     return data.get("data") or data
 
 
+async def user_eligibility_status(username: str) -> dict:
+    """用户补贴资格认证状态（PAT 鉴权）。"""
+    key = await _user_key(username)
+    if not key:
+        return {"eligible": False, "status": "none", "record": None}
+    data = await _request(
+        method="GET", path="/api/subsidy/eligibility/status",
+        headers={"Authorization": f"Bearer {key}"},
+    )
+    return data.get("data") or data
+
+
+async def user_eligibility_apply(username: str, payload: dict) -> dict:
+    """提交补贴资格认证申请（PAT 鉴权）。"""
+    key = await _user_key(username)
+    if not key:
+        return {"success": False, "error": "用户未绑定算力账号"}
+    data = await _request(
+        method="POST", path="/api/subsidy/eligibility/apply",
+        headers={"Authorization": f"Bearer {key}"}, json=payload,
+    )
+    return data.get("data") or data
+
+
+async def user_subsidy_apply(username: str, period: str = "", note: str = "") -> dict:
+    """手动申请补贴（PAT 鉴权，需资格认证通过）。"""
+    key = await _user_key(username)
+    if not key:
+        return {"success": False, "error": "用户未绑定算力账号"}
+    data = await _request(
+        method="POST", path="/api/subsidy/apply",
+        headers={"Authorization": f"Bearer {key}"},
+        json={"period": period, "note": note},
+    )
+    return data.get("data") or data
+
+
+async def user_subsidy_calc_preview(username: str, period: str = "") -> dict:
+    """补贴金额预览（PAT 鉴权，不生成记录）。"""
+    key = await _user_key(username)
+    if not key:
+        return {"success": False, "error": "用户未绑定算力账号"}
+    qs = f"?period={period}" if period else ""
+    data = await _request(
+        method="GET", path=f"/api/subsidy/calc-preview{qs}",
+        headers={"Authorization": f"Bearer {key}"},
+    )
+    return data.get("data") or data
+
+
 async def delete_token(token_id: int) -> dict:
     """删除引擎令牌。"""
     return await _request(