# store

**Repository Path**: uniquehwx/store

## Basic Information

- **Project Name**: store
- **Description**: 储能项目学习
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-04-02
- **Last Updated**: 2026-04-09

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 储能管理系统

一个基于FastAPI的储能管理系统，支持电价波动模拟、电池SOC跟踪、策略引擎和收益评估。

## 功能特性

### 核心功能
- **电价波动模拟**: 基于时间的峰谷电价生成，支持随机波动
- **电池SOC计算**: 实时计算电池电量百分比，记录SOC历史
- **安全限制屏障**: 充放电功率限制、SOC边界保护
- **策略引擎**: 支持多种充放电策略（峰谷套利、时间策略）
- **收益评估**: 充放电收益计算、交易记录、收益分析

### API端点

#### 储能相关
- `GET /api/storages/{id}/soc` - 获取当前SOC
- `GET /api/storages/{id}/soc-history` - 获取SOC历史

#### 电价相关
- `GET /api/prices/real-time` - 获取实时电价
- `GET /api/prices/history` - 获取电价历史
- `GET /api/prices/current-period` - 获取当前电价时段

#### 安全相关
- `GET /api/security/limits/{storage_id}` - 获取安全限制配置
- `POST /api/security/validate/{storage_id}` - 验证操作安全性

#### 策略相关
- `POST /api/strategy/execute/{storage_id}` - 执行策略
- `GET /api/strategy/history/{storage_id}` - 获取策略历史
- `GET /api/strategies` - 获取所有已注册策略

#### 收益相关
- `GET /api/revenue/{storage_id}` - 获取收益汇总
- `GET /api/revenue/{storage_id}/date/{date}` - 获取指定日期收益
- `GET /api/revenue/{storage_id}/month` - 获取指定月份收益
- `GET /api/revenue/{storage_id}/history` - 获取交易历史

## 快速开始

### 安装依赖

```bash
pip install -r requirements.txt
```

### 启动服务器

```bash
uvicorn main:app --reload --host 0.0.0.0 --port 8000
```

### 访问API文档

- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc

## 使用示例

### 1. 创建储能单元

```bash
curl -X POST "http://localhost:8000/api/storages" \
  -H "Content-Type: application/json" \
  -d '{
    "id": "storage-001",
    "name": "Building A Storage",
    "capacity": 100.0,
    "current_level": 50.0,
    "location": "Building A",
    "status": "idle"
  }'
```

### 2. 获取当前电价

```bash
curl "http://localhost:8000/api/prices/real-time"
```

### 3. 获取当前SOC

```bash
curl "http://localhost:8000/api/storages/storage-001/soc"
```

### 4. 执行策略

```bash
curl -X POST "http://localhost:8000/api/strategy/execute/storage-001"
```

### 5. 获取收益汇总

```bash
curl "http://localhost:8000/api/revenue/storage-001"
```

## 项目结构

```
储能/
├── app/
│   ├── models/              # Pydantic模型
│   │   ├── storage.py       # 储能相关模型
│   │   ├── pricing.py       # 电价模型
│   │   ├── security.py      # 安全限制模型
│   │   ├── strategy.py      # 策略模型
│   │   └── revenue.py       # 收益模型
│   ├── services/            # 业务逻辑服务
│   │   ├── storage_service.py       # 储能服务
│   │   ├── pricing_service.py       # 电价服务
│   │   ├── security_service.py      # 安全服务
│   │   ├── strategy_service.py      # 策略服务
│   │   └── revenue_service.py       # 收益服务
│   ├── database/            # 数据库
│   │   └── db.py            # SQLite数据库初始化
│   └── main.py              # FastAPI应用入口
├── tests/
│   └── test_core.py         # 测试文件
├── requirements.txt         # Python依赖
└── README.md               # 项目文档
```

## 技术栈

- **FastAPI**: Web框架
- **Pydantic**: 数据验证
- **SQLAlchemy**: ORM（数据库）
- **SQLite**: 嵌入式数据库
- **Python 3.9+**: 编程语言

## 策略说明

### 峰谷套利策略

- **充电条件**: SOC低于max_soc且电价低于charge_threshold
- **放电条件**: SOC高于min_soc且电价高于discharge_threshold
- **默认参数**:
  - charge_threshold: 0.4 (40%)
  - discharge_threshold: 0.6 (60%)
  - min_soc: 0.1 (10%)
  - max_soc: 0.9 (90%)

### 简单时间策略

- **充电视段**: 23:00 - 7:00 (夜间)
- **放电视段**: 8:00 - 12:00 (白天)

## 运行测试

```bash
pytest tests/test_core.py -v
```

## 安全限制

### 默认安全限制

- **最大充电功率**: 50 kW
- **最大放电功率**: 50 kW
- **最大充电SOC**: 95%
- **最小放电SOC**: 5%

可以通过修改`SecurityService`的配置来调整这些限制。

## 数据库

系统使用SQLite数据库进行数据持久化。数据库文件为`energy_storage.db`。

### 数据库表

- **storage_units**: 储能单元信息
- **soc_records**: SOC历史记录
- **price_history**: 电价历史
- **revenue_records**: 收益记录

## 扩展开发

### 添加新策略

继承`ChargingStrategy`基类并实现`evaluate`方法：

```python
class MyStrategy(ChargingStrategy):
    def evaluate(self, current_soc, current_price, price_model):
        # 实现策略逻辑
        if condition:
            return "charge"
        elif another_condition:
            return "discharge"
        else:
            return "idle"

    def get_parameters(self):
        return {"name": "My Strategy", ...}

# 注册策略
strategy_service.register_strategy(MyStrategy())
```

### 添加新API端点

在`main.py`中添加新的路由：

```python
@app.get("/api/your-endpoint")
async def your_endpoint(db: Session = get_db()):
    # 实现逻辑
    return {"data": ...}
```

## 许可证

MIT License