diff --git a/danding_bot/plugins/danding_points/README.md b/danding_bot/plugins/danding_points/README.md
new file mode 100644
index 0000000..6e4843c
--- /dev/null
+++ b/danding_bot/plugins/danding_points/README.md
@@ -0,0 +1,205 @@
+# Danding Points 插件
+
+全局积分/虚拟货币服务层，为其他插件提供统一的积分管理能力。用户在一个插件中获得的积分可以在另一个插件中消费。
+
+本插件不包含任何用户交互命令（无 NoneBot matcher），纯 API 服务层，供其他插件直接 import 调用。
+
+## 目录结构
+
+```
+danding_points/
+├── __init__.py     # 插件元数据 & points_api 单例导出
+├── config.py       # 配置类
+├── database.py     # SQLite 数据库操作
+└── api.py          # PointsAPI 核心
+```
+
+## 配置选项
+
+在 `.env` 文件或 NoneBot 配置文件中添加（均带 `DANDING_` 前缀）：
+
+| 配置项 | 类型 | 默认值 | 说明 |
+|--------|------|--------|------|
+| `DANDING_POINTS_DB_FILE` | str | `data/danding_points/points.db` | 数据库文件路径 |
+| `DANDING_POINTS_MAX_BALANCE` | int | `0` | 用户积分余额上限，`0` = 无限制 |
+| `DANDING_POINTS_MAX_PER_OPERATION` | int | `0` | 单次操作积分上限，`0` = 无限制 |
+| `DANDING_POINTS_LOG_RETENTION_DAYS` | int | `365` | 流水日志保留天数 |
+
+## API 接口
+
+所有方法均为 `async`，通过 `points_api` 单例调用。
+
+### `get_balance(user_id: str) -> int`
+
+查询用户积分余额。用户不存在时自动返回 `0`，不会创建账户。
+
+```python
+balance = await points_api.get_balance("123456")
+```
+
+### `add_points(user_id, amount, source, reason=None) -> Tuple[bool, int]`
+
+为用户增加积分。用户不存在时自动建户。
+
+- `amount`: 正整数，必须 > 0
+- `source`: 来源标识，不能为空（用于流水追踪）
+- `reason`: 变动原因，可选
+- 返回 `(success, new_balance)`，失败时 `success=False`
+
+触发校验：操作上限 → 余额上限。任一校验失败均返回 `(False, 当前余额)`。
+
+```python
+ok, balance = await points_api.add_points("123456", 100, "gacha", "抽卡奖励")
+if ok:
+    print(f"充值成功，当前余额: {balance}")
+```
+
+### `spend_points(user_id, amount, source, reason=None) -> Tuple[bool, int]`
+
+消费用户积分。余额不足时返回 `(False, 当前余额)`。
+
+- `amount`: 正整数，必须 > 0
+- 流水中 `amount` 记录为负数
+
+```python
+ok, balance = await points_api.spend_points("123456", 30, "shop", "购买道具")
+if not ok:
+    print("余额不足")
+```
+
+### `set_points(user_id, amount, source, reason=None) -> Tuple[bool, int]`
+
+直接设定用户积分。**绕过** 操作上限和余额上限约束。
+
+- `amount`: 非负整数，必须 >= 0
+- 新值等于旧值时不写流水，直接返回 `(True, amount)`
+- `total_earned` 仅累加正向差额（设低不影响）
+
+```python
+ok, balance = await points_api.set_points("123456", 500, "admin", "管理员调整")
+```
+
+### `get_transactions(user_id, limit=20, offset=0) -> List[dict]`
+
+查询用户积分流水记录，按时间倒序。
+
+- `limit`: 1~100，超出范围自动裁剪
+- `offset`: 分页偏移量，>= 0
+
+返回字段：`id, user_id, amount, balance_after, source, reason, created_at`
+
+```python
+txs = await points_api.get_transactions("123456", limit=10, offset=0)
+for tx in txs:
+    print(f"{tx['created_at']}  {tx['amount']:+d}  余额:{tx['balance_after']}")
+```
+
+### `get_ranking(limit=10, order_by="points") -> List[dict]`
+
+查询积分排行榜。
+
+- `limit`: 1~100
+- `order_by`: `"points"`（按余额）或 `"total_earned"`（按累计获得），其他值回退为 `"points"`
+- 使用 `RANK()` 窗口函数，同分并列，次级排序按 `user_id` 字母序
+
+返回字段：`rank, user_id, points, total_earned, total_spent`
+
+```python
+ranking = await points_api.get_ranking(limit=10, order_by="points")
+for r in ranking:
+    print(f"#{r['rank']}  {r['user_id']}  {r['points']}分")
+```
+
+## 其他插件对接
+
+### 基本用法
+
+在需要积分功能的插件中 import 单例即可：
+
+```python
+from nonebot import require
+
+require("danding_points")
+from danding_bot.plugins.danding_points import points_api
+```
+
+### 示例：抽卡插件发放奖励
+
+```python
+# 在 onmyoji_gacha 插件中
+from danding_bot.plugins.danding_points import points_api
+
+async def reward_user(user_id: str):
+    ok, balance = await points_api.add_points(
+        user_id, 50, "onmyoji_gacha", "SSR 抽到奖励"
+    )
+    return balance
+```
+
+### 示例：商店插件消费积分
+
+```python
+# 在 shop 插件中
+from danding_bot.plugins.danding_points import points_api
+
+async def buy_item(user_id: str, cost: int):
+    ok, balance = await points_api.spend_points(
+        user_id, cost, "shop", "购买商品"
+    )
+    if not ok:
+        return "积分不足"
+    return f"购买成功，剩余 {balance} 积分"
+```
+
+### 示例：管理员插件调整积分
+
+```python
+# 在 danding_api 插件中
+from danding_bot.plugins.danding_points import points_api
+
+async def admin_set(user_id: str, amount: int):
+    ok, balance = await points_api.set_points(
+        user_id, amount, "danding_api", "管理员手动调整"
+    )
+    return balance
+```
+
+### source 命名建议
+
+`source` 参数用于标识积分变动来源，建议各插件使用自身插件名作为 source：
+
+| 插件 | source 值 |
+|------|-----------|
+| onmyoji_gacha | `"onmyoji_gacha"` |
+| danding_api | `"danding_api"` |
+| shop | `"shop"` |
+| sign_in | `"sign_in"` |
+
+## 数据库
+
+使用 SQLite，数据文件位于 `data/danding_points/points.db`，无需额外配置。
+
+### 表结构
+
+**user_points** — 用户积分账户
+
+| 字段 | 类型 | 说明 |
+|------|------|------|
+| user_id | TEXT PK | 用户 ID |
+| points | INTEGER | 当前余额，>= 0 |
+| total_earned | INTEGER | 累计获得 |
+| total_spent | INTEGER | 累计消费 |
+| created_at | TEXT | 创建时间 |
+| updated_at | TEXT | 最后更新时间 |
+
+**point_transactions** — 积分变动流水
+
+| 字段 | 类型 | 说明 |
+|------|------|------|
+| id | INTEGER PK | 自增 ID |
+| user_id | TEXT | 用户 ID |
+| amount | INTEGER | 变动数额（消费为负） |
+| balance_after | INTEGER | 变动后余额 |
+| source | TEXT | 来源标识 |
+| reason | TEXT | 变动原因 |
+| created_at | TEXT | 创建时间 |
diff --git a/danding_bot/plugins/danding_points/__init__.py b/danding_bot/plugins/danding_points/__init__.py
new file mode 100644
index 0000000..3962ee2
--- /dev/null
+++ b/danding_bot/plugins/danding_points/__init__.py
@@ -0,0 +1,18 @@
+from nonebot.plugin import PluginMetadata
+from .config import Config
+from .api import PointsAPI
+
+__plugin_meta__ = PluginMetadata(
+    name="Danding Points",
+    description="Global points/virtual currency system for danding-bot",
+    usage="Import points_api from this plugin to use the points API",
+    type="service",
+    homepage="https://github.com/danding-bot/danding-bot",
+    config=Config,
+)
+
+# Initialize configuration and API
+config = Config()
+points_api = PointsAPI(config)
+
+__all__ = ["points_api", "PointsAPI", "Config"]
diff --git a/danding_bot/plugins/danding_points/api.py b/danding_bot/plugins/danding_points/api.py
new file mode 100644
index 0000000..f649c26
--- /dev/null
+++ b/danding_bot/plugins/danding_points/api.py
@@ -0,0 +1,294 @@
+import asyncio
+import threading
+from datetime import datetime
+from typing import Tuple, List, Dict, Any
+from .config import Config
+from .database import PointsDatabase
+
+
+class PointsAPI:
+    """Points system API for managing user points."""
+
+    def __init__(self, config: Config):
+        self.config = config
+        self.db = PointsDatabase(config)
+        self._lock = threading.Lock()
+
+    async def get_balance(self, user_id: str) -> int:
+        """Get user's current points balance."""
+        return await asyncio.to_thread(self.db.get_user_balance, user_id)
+
+    async def add_points(
+        self, user_id: str, amount: int, source: str, reason: str = None
+    ) -> Tuple[bool, int]:
+        """Add points to user account.
+
+        Returns: (success, new_balance)
+        """
+        # Parameter validation
+        if not isinstance(amount, int) or amount <= 0:
+            return False, 0
+        if not user_id or not source:
+            return False, 0
+
+        # Operation limit validation
+        if self.config.POINTS_MAX_PER_OPERATION > 0:
+            if amount > self.config.POINTS_MAX_PER_OPERATION:
+                return False, 0
+
+        def _add():
+            with self._lock:
+                conn = self.db.get_connection()
+                cursor = conn.cursor()
+                try:
+                    # Ensure user exists
+                    self.db.ensure_user_exists(user_id)
+
+                    # Get current balance
+                    cursor.execute(
+                        "SELECT points FROM user_points WHERE user_id = ?",
+                        (user_id,),
+                    )
+                    row = cursor.fetchone()
+                    current_balance = row["points"] if row else 0
+
+                    # Check balance limit
+                    new_balance = current_balance + amount
+                    if self.config.POINTS_MAX_BALANCE > 0:
+                        if new_balance > self.config.POINTS_MAX_BALANCE:
+                            conn.close()
+                            return False, current_balance
+
+                    # Update balance and total_earned
+                    now = datetime.now().isoformat()
+                    cursor.execute(
+                        """
+                        UPDATE user_points
+                        SET points = ?, total_earned = total_earned + ?, updated_at = ?
+                        WHERE user_id = ?
+                        """,
+                        (new_balance, amount, now, user_id),
+                    )
+
+                    # Write transaction log
+                    cursor.execute(
+                        """
+                        INSERT INTO point_transactions
+                        (user_id, amount, balance_after, source, reason, created_at)
+                        VALUES (?, ?, ?, ?, ?, ?)
+                        """,
+                        (user_id, amount, new_balance, source, reason, now),
+                    )
+
+                    conn.commit()
+                    conn.close()
+                    return True, new_balance
+                except Exception:
+                    conn.close()
+                    return False, 0
+
+        return await asyncio.to_thread(_add)
+
+    async def spend_points(
+        self, user_id: str, amount: int, source: str, reason: str = None
+    ) -> Tuple[bool, int]:
+        """Spend points from user account.
+
+        Returns: (success, new_balance)
+        """
+        # Parameter validation
+        if not isinstance(amount, int) or amount <= 0:
+            return False, 0
+        if not user_id or not source:
+            return False, 0
+
+        # Operation limit validation
+        if self.config.POINTS_MAX_PER_OPERATION > 0:
+            if amount > self.config.POINTS_MAX_PER_OPERATION:
+                return False, 0
+
+        def _spend():
+            with self._lock:
+                conn = self.db.get_connection()
+                cursor = conn.cursor()
+                try:
+                    # Ensure user exists
+                    self.db.ensure_user_exists(user_id)
+
+                    # Get current balance
+                    cursor.execute(
+                        "SELECT points FROM user_points WHERE user_id = ?",
+                        (user_id,),
+                    )
+                    row = cursor.fetchone()
+                    current_balance = row["points"] if row else 0
+
+                    # Check sufficient balance
+                    if current_balance < amount:
+                        conn.close()
+                        return False, current_balance
+
+                    # Update balance and total_spent
+                    new_balance = current_balance - amount
+                    now = datetime.now().isoformat()
+                    cursor.execute(
+                        """
+                        UPDATE user_points
+                        SET points = ?, total_spent = total_spent + ?, updated_at = ?
+                        WHERE user_id = ?
+                        """,
+                        (new_balance, amount, now, user_id),
+                    )
+
+                    # Write transaction log (amount as negative)
+                    cursor.execute(
+                        """
+                        INSERT INTO point_transactions
+                        (user_id, amount, balance_after, source, reason, created_at)
+                        VALUES (?, ?, ?, ?, ?, ?)
+                        """,
+                        (user_id, -amount, new_balance, source, reason, now),
+                    )
+
+                    conn.commit()
+                    conn.close()
+                    return True, new_balance
+                except Exception:
+                    conn.close()
+                    return False, 0
+
+        return await asyncio.to_thread(_spend)
+
+    async def set_points(
+        self, user_id: str, amount: int, source: str, reason: str = None
+    ) -> Tuple[bool, int]:
+        """Set user's points to exact amount.
+
+        Returns: (success, new_balance)
+        """
+        # Parameter validation
+        if not isinstance(amount, int) or amount < 0:
+            return False, 0
+        if not user_id or not source:
+            return False, 0
+
+        def _set():
+            with self._lock:
+                conn = self.db.get_connection()
+                cursor = conn.cursor()
+                try:
+                    # Ensure user exists
+                    self.db.ensure_user_exists(user_id)
+
+                    # Get current balance
+                    cursor.execute(
+                        "SELECT points, total_earned FROM user_points WHERE user_id = ?",
+                        (user_id,),
+                    )
+                    row = cursor.fetchone()
+                    current_balance = row["points"] if row else 0
+                    current_earned = row["total_earned"] if row else 0
+
+                    # If new value equals old value, return without writing
+                    if current_balance == amount:
+                        conn.close()
+                        return True, amount
+
+                    # Calculate difference for total_earned (only positive diff)
+                    diff = amount - current_balance
+                    earned_diff = max(0, diff)
+
+                    # Update balance and total_earned
+                    now = datetime.now().isoformat()
+                    cursor.execute(
+                        """
+                        UPDATE user_points
+                        SET points = ?, total_earned = total_earned + ?, updated_at = ?
+                        WHERE user_id = ?
+                        """,
+                        (amount, earned_diff, now, user_id),
+                    )
+
+                    # Write transaction log
+                    cursor.execute(
+                        """
+                        INSERT INTO point_transactions
+                        (user_id, amount, balance_after, source, reason, created_at)
+                        VALUES (?, ?, ?, ?, ?, ?)
+                        """,
+                        (user_id, diff, amount, source, reason, now),
+                    )
+
+                    conn.commit()
+                    conn.close()
+                    return True, amount
+                except Exception:
+                    conn.close()
+                    return False, 0
+
+        return await asyncio.to_thread(_set)
+
+    async def get_transactions(
+        self, user_id: str, limit: int = 20, offset: int = 0
+    ) -> List[Dict[str, Any]]:
+        """Get transaction history for a user.
+
+        Returns: List of transaction dicts
+        """
+        # Normalize parameters
+        limit = max(1, min(100, limit))
+        offset = max(0, offset)
+
+        def _get():
+            conn = self.db.get_connection()
+            cursor = conn.cursor()
+            cursor.execute(
+                """
+                SELECT id, user_id, amount, balance_after, source, reason, created_at
+                FROM point_transactions
+                WHERE user_id = ?
+                ORDER BY id DESC
+                LIMIT ? OFFSET ?
+                """,
+                (user_id, limit, offset),
+            )
+            rows = cursor.fetchall()
+            conn.close()
+            return [dict(row) for row in rows]
+
+        return await asyncio.to_thread(_get)
+
+    async def get_ranking(
+        self, limit: int = 10, order_by: str = "points"
+    ) -> List[Dict[str, Any]]:
+        """Get points ranking.
+
+        Returns: List of ranking dicts with rank field
+        """
+        # Normalize parameters
+        limit = max(1, min(100, limit))
+        if order_by not in ("points", "total_earned"):
+            order_by = "points"
+
+        def _get():
+            conn = self.db.get_connection()
+            cursor = conn.cursor()
+
+            order_column = "points" if order_by == "points" else "total_earned"
+            query = f"""
+                SELECT
+                    RANK() OVER (ORDER BY {order_column} DESC) as rank,
+                    user_id,
+                    points,
+                    total_earned,
+                    total_spent
+                FROM user_points
+                ORDER BY {order_column} DESC, user_id ASC
+                LIMIT ?
+            """
+            cursor.execute(query, (limit,))
+            rows = cursor.fetchall()
+            conn.close()
+            return [dict(row) for row in rows]
+
+        return await asyncio.to_thread(_get)
diff --git a/danding_bot/plugins/danding_points/config.py b/danding_bot/plugins/danding_points/config.py
new file mode 100644
index 0000000..33025b2
--- /dev/null
+++ b/danding_bot/plugins/danding_points/config.py
@@ -0,0 +1,27 @@
+from pydantic import BaseSettings, validator
+from pathlib import Path
+
+
+class Config(BaseSettings):
+    """Points system configuration."""
+
+    POINTS_DB_FILE: str = "data/danding_points/points.db"
+    POINTS_MAX_BALANCE: int = 0  # 0 = unlimited
+    POINTS_MAX_PER_OPERATION: int = 0  # 0 = unlimited
+    POINTS_LOG_RETENTION_DAYS: int = 365
+
+    class Config:
+        env_prefix = "DANDING_"
+        case_sensitive = True
+
+    @validator("POINTS_MAX_BALANCE", "POINTS_MAX_PER_OPERATION", "POINTS_LOG_RETENTION_DAYS")
+    def validate_non_negative(cls, v):
+        if v < 0:
+            raise ValueError("Value must be non-negative")
+        return v
+
+    @validator("POINTS_DB_FILE")
+    def validate_db_path(cls, v):
+        if not v:
+            raise ValueError("Database file path cannot be empty")
+        return v
diff --git a/danding_bot/plugins/danding_points/database.py b/danding_bot/plugins/danding_points/database.py
new file mode 100644
index 0000000..44b05e4
--- /dev/null
+++ b/danding_bot/plugins/danding_points/database.py
@@ -0,0 +1,100 @@
+import sqlite3
+import os
+from datetime import datetime
+from typing import Optional, List, Dict, Any
+from .config import Config
+
+
+class PointsDatabase:
+    """SQLite database handler for points system."""
+
+    def __init__(self, config: Config):
+        self.config = config
+        self.db_path = config.POINTS_DB_FILE
+        self._ensure_db_dir()
+        self._init_db()
+
+    def _ensure_db_dir(self):
+        """Create database directory if it doesn't exist."""
+        db_dir = os.path.dirname(self.db_path)
+        if db_dir:
+            os.makedirs(db_dir, exist_ok=True)
+
+    def _init_db(self):
+        """Initialize database tables."""
+        conn = sqlite3.connect(self.db_path, timeout=5.0)
+        cursor = conn.cursor()
+
+        # Create user_points table
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS user_points (
+                user_id TEXT PRIMARY KEY,
+                points INTEGER NOT NULL DEFAULT 0 CHECK(points >= 0),
+                total_earned INTEGER NOT NULL DEFAULT 0,
+                total_spent INTEGER NOT NULL DEFAULT 0,
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+            )
+            """
+        )
+
+        # Create point_transactions table
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS point_transactions (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                user_id TEXT NOT NULL,
+                amount INTEGER NOT NULL,
+                balance_after INTEGER NOT NULL,
+                source TEXT NOT NULL,
+                reason TEXT,
+                created_at TEXT NOT NULL
+            )
+            """
+        )
+
+        # Create indexes
+        cursor.execute(
+            "CREATE INDEX IF NOT EXISTS idx_transactions_user_id ON point_transactions(user_id)"
+        )
+        cursor.execute(
+            "CREATE INDEX IF NOT EXISTS idx_transactions_source ON point_transactions(source)"
+        )
+        cursor.execute(
+            "CREATE INDEX IF NOT EXISTS idx_transactions_created_at ON point_transactions(created_at)"
+        )
+
+        conn.commit()
+        conn.close()
+
+    def get_connection(self) -> sqlite3.Connection:
+        """Get a database connection."""
+        conn = sqlite3.connect(self.db_path, timeout=5.0)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def get_user_balance(self, user_id: str) -> int:
+        """Get user's current points balance."""
+        conn = self.get_connection()
+        cursor = conn.cursor()
+        cursor.execute("SELECT points FROM user_points WHERE user_id = ?", (user_id,))
+        row = cursor.fetchone()
+        conn.close()
+        return row["points"] if row else 0
+
+    def ensure_user_exists(self, user_id: str) -> None:
+        """Create user account if it doesn't exist."""
+        conn = self.get_connection()
+        cursor = conn.cursor()
+        now = datetime.now().isoformat()
+        cursor.execute(
+            """
+            INSERT OR IGNORE INTO user_points
+            (user_id, points, total_earned, total_spent, created_at, updated_at)
+            VALUES (?, 0, 0, 0, ?, ?)
+            """,
+            (user_id, now, now),
+        )
+        conn.commit()
+        conn.close()
diff --git a/test_danding_points.py b/test_danding_points.py
new file mode 100644
index 0000000..3de2a9f
--- /dev/null
+++ b/test_danding_points.py
@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""Manual test script for danding_points plugin."""
+
+import asyncio
+import sys
+from pathlib import Path
+
+# Add project root to path
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+from danding_bot.plugins.danding_points import points_api
+
+
+async def test_basic_operations():
+    """Test basic points operations."""
+    print("Testing basic operations...")
+
+    # Test 1: Get balance for non-existent user
+    balance = await points_api.get_balance("test_user_1")
+    assert balance == 0, f"Expected 0, got {balance}"
+    print("✓ Non-existent user returns 0 balance")
+
+    # Test 2: Add points (auto-create user)
+    success, new_balance = await points_api.add_points(
+        "test_user_1", 100, "test_source", "test reason"
+    )
+    assert success and new_balance == 100, f"Add failed: {success}, {new_balance}"
+    print("✓ Add points works and auto-creates user")
+
+    # Test 3: Get balance after add
+    balance = await points_api.get_balance("test_user_1")
+    assert balance == 100, f"Expected 100, got {balance}"
+    print("✓ Balance updated correctly")
+
+    # Test 4: Spend points
+    success, new_balance = await points_api.spend_points(
+        "test_user_1", 30, "test_source", "spend reason"
+    )
+    assert success and new_balance == 70, f"Spend failed: {success}, {new_balance}"
+    print("✓ Spend points works")
+
+    # Test 5: Spend more than balance (should fail)
+    success, new_balance = await points_api.spend_points(
+        "test_user_1", 100, "test_source", "should fail"
+    )
+    assert not success, "Should fail when spending more than balance"
+    print("✓ Spend fails when insufficient balance")
+
+    # Test 6: Set points
+    success, new_balance = await points_api.set_points(
+        "test_user_1", 50, "test_source", "set reason"
+    )
+    assert success and new_balance == 50, f"Set failed: {success}, {new_balance}"
+    print("✓ Set points works")
+
+    # Test 7: Get transactions
+    transactions = await points_api.get_transactions("test_user_1", limit=10)
+    assert len(transactions) > 0, "Should have transactions"
+    print(f"✓ Get transactions works ({len(transactions)} transactions)")
+
+    # Test 8: Get ranking
+    ranking = await points_api.get_ranking(limit=10)
+    assert len(ranking) > 0, "Should have ranking entries"
+    assert "rank" in ranking[0], "Ranking should have rank field"
+    print(f"✓ Get ranking works ({len(ranking)} entries)")
+
+    print("\n✅ All tests passed!")
+
+
+if __name__ == "__main__":
+    asyncio.run(test_basic_operations())