光伏资产运营平台和传统监控平台有什么区别?

分布式光伏投资公司资产管理后端用 Django + Celery + Redis,完整功能 + 异步任务 + 缓存。

一、整体架构

graph LR
    A["Django REST API"] --> B["PostgreSQL"]
    A --> C["Redis(缓存)"]
    A --> D["Celery 异步任务"]
    D --> E["定时任务(beat)"]
    D --> F["监控数据同步"]
    D --> G["报表生成"]

二、Django Model

from django.db import models

class Investor(models.Model):
    name = models.CharField(max_length=128)
    type = models.CharField(max_length=32, choices=[("corp", "企业"), ("fund", "基金"), ("indiv", "个人")])
    total_investment = models.DecimalField(max_digits=15, decimal_places=2, default=0)

class Station(models.Model):
    investor = models.ForeignKey(Investor, on_delete=models.PROTECT, related_name="stations")
    name = models.CharField(max_length=128)
    capacity_kw = models.FloatField()
    location = models.JSONField()
    grid_connect_date = models.DateField()
    status = models.CharField(max_length=32, choices=[("planning", "规划"), ("construction", "建设中"), ("operating", "运营"), ("retired", "退役")])
    investment_yuan = models.DecimalField(max_digits=15, decimal_places=2)

class Contract(models.Model):
    station = models.ForeignKey(Station, on_delete=models.CASCADE)
    type = models.CharField(max_length=32)
    counterparty = models.CharField(max_length=128)
    start_date = models.DateField()
    end_date = models.DateField()
    amount_yuan = models.DecimalField(max_digits=15, decimal_places=2)

class Revenue(models.Model):
    station = models.ForeignKey(Station, on_delete=models.CASCADE)
    period_year = models.IntegerField()
    period_month = models.IntegerField()
    generation_kwh = models.FloatField()
    revenue_self = models.DecimalField(max_digits=15, decimal_places=2)
    revenue_sell = models.DecimalField(max_digits=15, decimal_places=2)
    expense = models.DecimalField(max_digits=15, decimal_places=2)
    net_profit = models.DecimalField(max_digits=15, decimal_places=2)
    class Meta:
        unique_together = [("station", "period_year", "period_month")]

三、REST API

from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response

class StationViewSet(viewsets.ModelViewSet):
    queryset = Station.objects.all()
    serializer_class = StationSerializer

    @action(detail=True, methods=["get"])
    def kpis(self, request, pk=None):
        station = self.get_object()
        revenues = Revenue.objects.filter(station=station, period_year=request.query_params.get("year"))
        return Response({
            "total_generation_kwh": sum(r.generation_kwh for r in revenues),
            "total_revenue": sum(r.revenue_self + r.revenue_sell for r in revenues),
            "total_expense": sum(r.expense for r in revenues),
            "net_profit": sum(r.net_profit for r in revenues),
        })

    @action(detail=True, methods=["get"])
    def irr(self, request, pk=None):
        return Response(calculate_irr(self.get_object()))

四、Redis 缓存

from django_redis import get_redis_connection
import json

class CachedStationService:
    @staticmethod
    def get_realtime_kpis(station_id: str):
        redis_client = get_redis_connection("default")
        cache_key = f"station:realtime:{station_id}"

        # 尝试缓存
        cached = redis_client.get(cache_key)
        if cached:
            return json.loads(cached)

        # 计算
        kpis = calculate_realtime_kpis(station_id)
        redis_client.setex(cache_key, 60, json.dumps(kpis))  # 60s TTL
        return kpis

五、Celery 异步任务

# tasks.py
from celery import shared_task
import httpx

@shared_task(bind=True, max_retries=3)
def sync_station_monthly_data(self, station_id: str, year: int, month: int):
    """从监控平台拉取月度数据"""
    try:
        station = Station.objects.get(pk=station_id)
        # 调用外部 API
        resp = httpx.post(
            "https://monitor.example.com/api/monthly",
            json={"station_code": station.code, "year": year, "month": month},
            timeout=30
        )
        data = resp.json()

        Revenue.objects.update_or_create(
            station=station,
            period_year=year,
            period_month=month,
            defaults={
                "generation_kwh": data["total_kwh"],
                "revenue_self": data["self_use_revenue"],
                "revenue_sell": data["sell_revenue"],
                "expense": data["expense"],
                "net_profit": data["net_profit"],
            }
        )
    except Exception as e:
        raise self.retry(exc=e, countdown=60)

@shared_task
def generate_monthly_reports():
    """月度报表生成(给所有投资人)"""
    for investor in Investor.objects.all():
        report = build_monthly_report(investor)
        send_email_report(investor.email, report)

六、定时任务(Celery Beat)

# celery.py
from celery import Celery
from celery.schedules import crontab

app = Celery('asset_mgmt')
app.config_from_object('django.conf:settings', namespace='CELERY')

app.conf.beat_schedule = {
    'sync-monthly-data': {
        'task': 'stations.tasks.sync_all_stations_monthly',
        'schedule': crontab(minute=0, hour=2, day_of_month=1),  # 每月 1 号 凌晨 2 点
    },
    'generate-monthly-reports': {
        'task': 'stations.tasks.generate_monthly_reports',
        'schedule': crontab(minute=0, hour=8, day_of_month=2),
    },
    'cache-realtime-data': {
        'task': 'stations.tasks.refresh_realtime_cache',
        'schedule': crontab(minute='*/5'),  # 每 5 分钟
    },
}

七、IRR + NPV 计算

import numpy_financial as npf

def calculate_irr(station):
    cashflows = [-float(station.investment_yuan)]
    for year in range(25):
        annual = Revenue.objects.filter(
            station=station,
            period_year=2025 + year
        ).aggregate(total=models.Sum("net_profit"))["total"] or 0
        cashflows.append(float(annual))

    return {
        "irr_pct": npf.irr(cashflows) * 100,
        "npv_8pct": npf.npv(0.08, cashflows),
        "payback_years": calculate_payback(cashflows),
    }

八、Django Admin(管理端)

from django.contrib import admin

@admin.register(Station)
class StationAdmin(admin.ModelAdmin):
    list_display = ["name", "capacity_kw", "investor", "status", "irr_pct"]
    list_filter = ["status", "investor"]
    search_fields = ["name", "code"]

    def irr_pct(self, obj):
        return f"{calculate_irr(obj)['irr_pct']:.1f}%"
    irr_pct.short_description = "IRR"

九、ZenovaOS 实践

ZenovaOS 给投资公司提供资产管理模块,集成 Django + Celery + Redis 全栈。

总结

光伏资产管理后端用 Django + Celery + Redis + PostgreSQL,API + 缓存 + 异步任务 + 报表完整方案。