DeepSeek V4 Official Release Deep Dive: MoE Sparse Attention + DSpark Speculative Decoding + Peak-Valley Pricing Economics

Core Insight: On June 29, 2026, DeepSeek announced the V4 official release for mid-July, with a simultaneous API peak-valley pricing mechanism — peak hours (9-12 AM, 2-6 PM) double the price. This is not a simple price hike, but a landmark shift in AI cloud services from “coarse supply” to “fine-grained operations.” Technically, DSpark speculative decoding boosts Flash generation speed by 85%, while DSA sparse attention compresses million-token inference computation to 27% of V3.2. The 1.6T-parameter MoE giant is using “technological leverage” to drive a dual revolution in both engineering and business models.


I. Background: From “Price Killer” to “Peak-Valley Pricing”

On April 24, 2026, DeepSeek V4 preview shocked the industry with ultra-low prices (Pro output at 6 RMB/million tokens, 1/17th of GPT-4o), earning the nickname “price killer.” In two months of gray-scale testing, V4 Flash alone reached 4.66 trillion tokens of weekly call volume, causing peak-hour interface timeouts.

This “growing pain” gave birth to peak-valley pricing — not a pure price increase, but resource optimization through price leverage:

              Peak Hour Compute Supply-Demand Gap
                           │
              ┌────────────┴────────────┐
              │                         │
        4.66T Tokens/Week         300% Timeout
        Call Volume                Rate Increase
              │                         │
              └────────────┬────────────┘
                           │
                    ┌──────▼──────┐
                    │ Peak-Valley │
                    │   Pricing   │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
         Peak Shaving  Guarantee    Guide Elastic
         Night Batch  Finance/Code  Offline Batch
        分流          保障          降价让利

The essence of peak-valley pricing is efficiency-oriented third-degree price discrimination:

  • Price-sensitive users (individual developers, nightly batch) → choose off-peak, costs halved
  • Time-sensitive users (financial trading, online services) → accept peak premium, service guaranteed
  • Strategic users (AI startups) → hybrid scheduling, total cost optimized

II. Model Architecture: 2nd Gen MoE + DSA Sparse Attention

Dual Version Matrix

┌─────────────────────────────────────────────────────────────┐
│                   DeepSeek V4 Model Matrix                   │
│                                                              │
│  ┌─────────────────────────┐  ┌─────────────────────────┐   │
│  │    V4 Pro (Flagship)    │  │   V4 Flash (Lightweight)│   │
│  ├─────────────────────────┤  ├─────────────────────────┤   │
│  │ Total params: 1.6T     │  │ Total params: 284B      │   │
│  │ Active params: 49B     │  │ Active params: 13B      │   │
│  │ Context: 1M tokens     │  │ Context: 1M tokens      │   │
│  │ Position: High perf    │  │ Position: High freq     │   │
│  │ Output: 6 RMB/M tokens │  │ Output: 2 RMB/M tokens  │   │
│  │ Peak price: 12 RMB     │  │ Peak price: 4 RMB       │   │
│  └─────────────────────────┘  └─────────────────────────┘   │
│                                                              │
│  Shared: MoE + DSA + 1M context + MIT license               │
└─────────────────────────────────────────────────────────────┘

DSA (Dense-Sparse Attention)

DSA compresses tokens based on importance scores, combining dense attention (global semantics) with sparse attention (local details). At 1M tokens, computation is just 27% of V3.2, with memory at 10%.

import torch
import torch.nn as nn
import torch.nn.functional as F
import math


class DSAAttention(nn.Module):
    """Dense-Sparse Attention mechanism"""
    
    def __init__(self, d_model, n_heads, compress_ratio=0.1, sparse_ratio=0.05):
        super().__init__()
        self.d_model = d_model
        self.n_heads = n_heads
        self.head_dim = d_model // n_heads
        self.compress_ratio = compress_ratio
        self.sparse_ratio = sparse_ratio
        
        self.wq = nn.Linear(d_model, d_model)
        self.wk = nn.Linear(d_model, d_model)
        self.wv = nn.Linear(d_model, d_model)
        self.wo = nn.Linear(d_model, d_model)
    
    def _compress_tokens(self, x: torch.Tensor) -> tuple:
        """Compress tokens by importance scoring"""
        batch_size, seq_len, _ = x.shape
        # Compute importance score (L2 norm per token)
        scores = torch.norm(x, dim=-1)  # [B, L]
        keep_len = max(1, int(seq_len * self.compress_ratio))
        
        # Top-k selection
        _, indices = torch.topk(scores, keep_len, dim=-1, sorted=False)
        indices = indices.sort(dim=-1).values  # maintain order
        
        # Gather compressed tokens
        compressed = torch.gather(
            x, 1, 
            indices.unsqueeze(-1).expand(-1, -1, self.d_model)
        )
        return compressed, indices
    
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        batch_size, seq_len, _ = x.shape
        
        # Project Q, K, V
        Q = self.wq(x).view(batch_size, seq_len, self.n_heads, self.head_dim)
        K = self.wk(x).view(batch_size, seq_len, self.n_heads, self.head_dim)
        V = self.wv(x).view(batch_size, seq_len, self.n_heads, self.head_dim)
        
        Q = Q.transpose(1, 2)
        K = K.transpose(1, 2)
        V = V.transpose(1, 2)
        
        # Dense path: compress K and V
        compressed_k, indices = self._compress_tokens(
            K.reshape(batch_size, seq_len, -1)
        )
        compressed_v, _ = self._compress_tokens(
            V.reshape(batch_size, seq_len, -1)
        )
        
        compressed_k = compressed_k.view(
            batch_size, -1, self.n_heads, self.head_dim
        ).transpose(1, 2)
        compressed_v = compressed_v.view(
            batch_size, -1, self.n_heads, self.head_dim
        ).transpose(1, 2)
        
        # Dense attention
        dense_scores = torch.matmul(Q, compressed_k.transpose(-2, -1))
        dense_scores = dense_scores / math.sqrt(self.head_dim)
        dense_attn = F.softmax(dense_scores, dim=-1)
        dense_out = torch.matmul(dense_attn, compressed_v)
        
        # Sparse path: only attend to locally important tokens
        sparse_mask = self._build_sparse_mask(seq_len, indices[0], 
                                               window=64, device=x.device)
        sparse_mask = sparse_mask.unsqueeze(0).unsqueeze(0)  # [1,1,L,L]
        
        sparse_scores = torch.matmul(Q, K.transpose(-2, -1))
        sparse_scores = sparse_scores / math.sqrt(self.head_dim)
        sparse_scores = sparse_scores.masked_fill(~sparse_mask, float('-inf'))
        sparse_attn = F.softmax(sparse_scores, dim=-1)
        sparse_out = torch.matmul(sparse_attn, V)
        
        # Fuse: 70% dense + 30% sparse
        output = 0.7 * dense_out + 0.3 * sparse_out
        
        output = output.transpose(1, 2).contiguous().view(
            batch_size, seq_len, -1
        )
        return self.wo(output)
    
    def _build_sparse_mask(self, seq_len, indices, window, device):
        """Build sparse attention mask"""
        mask = torch.zeros(seq_len, seq_len, dtype=torch.bool, device=device)
        indices_set = set(indices.tolist())
        for i in range(seq_len):
            start = max(0, i - window)
            end = min(seq_len, i + window + 1)
            for j in range(start, end):
                if j in indices_set:
                    mask[i, j] = True
        return mask

III. DSpark Speculative Decoding

DSpark uses a small draft model (V4 Flash, 13B) to generate candidate tokens, verified in parallel by the target model (V4 Pro, 49B).

DSpark Workflow:

Input: "DeepSeek V4's advantage is"
         │
         ▼
  ┌─────────────────┐     ┌─────────────────┐
  │ Draft Model     │     │ Target Model    │
  │ V4 Flash 13B    │     │ V4 Pro 49B      │
  │ Fast auto-reg   │     │ Parallel verify │
  └────────┬────────┘     └────────┬────────┘
           │                       │
  Generate candidates:     Parallel verify each:
  "its" "sparse" "attn"   ✅ ✅ ✅ All accepted
  1 step per token        4 tokens verified at once
           │                       │
           └───────────┬───────────┘
                       │
                       ▼
              ┌─────────────────┐
              │ Accept/Reject   │
              │ Accept rate ~80%│
              │ Speedup ~3.7x   │
              └─────────────────┘
                       │
                       ▼
             Output: "its sparse attention"
             Traditional: 4 steps → DSpark: 2 steps
             Flash 85% faster end-to-end ✅
import torch
import torch.nn.functional as F
from typing import List, Tuple


class DSparkDecoder:
    """
    Speculative decoding: draft generates, target verifies in parallel
    Flash version achieves 85% speed improvement
    """
    
    def __init__(self, draft_model, target_model, gamma=5, temperature=0.7):
        self.draft_model = draft_model
        self.target_model = target_model
        self.gamma = gamma
        self.temperature = temperature
    
    @torch.no_grad()
    def generate(self, input_ids, max_new_tokens=256):
        prefix = input_ids.clone()
        all_tokens = input_ids[0].tolist()
        draft_steps = 0
        target_steps = 0
        
        while len(all_tokens) < len(input_ids[0]) + max_new_tokens:
            # Draft: generate gamma candidate tokens
            draft_tokens = []
            current = prefix.clone()
            for _ in range(self.gamma):
                logits = self.draft_model(current)
                probs = F.softmax(logits[0, -1] / self.temperature, dim=-1)
                token = torch.multinomial(probs, 1).item()
                draft_tokens.append(token)
                current = torch.cat([current, torch.tensor([[token]], 
                    device=current.device)], dim=1)
            draft_steps += 1
            
            # Target: parallel verification
            verify_input = torch.cat([
                prefix,
                torch.tensor([draft_tokens], device=prefix.device)
            ], dim=1)
            target_logits = self.target_model(verify_input)
            target_steps += 1
            
            # Rejection sampling
            n_accepted = 0
            for i, dt in enumerate(draft_tokens):
                with torch.no_grad():
                    draft_logits = self.draft_model(
                        prefix if i == 0 else 
                        torch.cat([prefix, torch.tensor([draft_tokens[:i]], 
                            device=prefix.device)], dim=1)
                    )
                    draft_prob = F.softmax(
                        draft_logits[0, -1] / self.temperature, dim=-1
                    )[dt].item()
                
                target_prob = F.softmax(
                    target_logits[0, prefix.shape[1] + i] / self.temperature, 
                    dim=-1
                )[dt].item()
                
                if target_prob > draft_prob:
                    if torch.rand(1).item() <= target_prob / draft_prob:
                        n_accepted += 1
                    else:
                        break
                else:
                    n_accepted += 1
            
            # Accept first n_accepted tokens
            for i in range(n_accepted):
                prefix = torch.cat([prefix, 
                    torch.tensor([[draft_tokens[i]]], device=prefix.device)], dim=1)
                all_tokens.append(draft_tokens[i])
            
            # Bonus token if rejection occurred
            if n_accepted < self.gamma:
                bonus_logits = target_logits[0, prefix.shape[1] - 1]
                bonus_prob = F.softmax(bonus_logits / self.temperature, dim=-1)
                bonus_token = torch.multinomial(bonus_prob, 1).item()
                prefix = torch.cat([prefix, 
                    torch.tensor([[bonus_token]], device=prefix.device)], dim=1)
                all_tokens.append(bonus_token)
        
        return all_tokens

IV. Peak-Valley Pricing: Technical Economics

Price Gradient Timeline

Price (RMB/million output tokens)
         │
   12 ───┤■■■■■■■■■■■■■■■■■■■■  V4 Pro Peak
         │                      (9-12, 14-18)
    6 ───┤■■■■■■■■■■  V4 Pro Off-Peak
         │
    4 ───┤■■■■■■■     V4 Flash Peak
         │
    2 ───┤■■■■         V4 Flash Off-Peak
         │
         └───────────────────────────► Time
               0   6   9  12  14  18  24

Optimal Scheduling Strategy

from datetime import datetime, timedelta
from enum import Enum
from dataclasses import dataclass


class Tier(Enum):
    OFF_PEAK = 0
    PEAK = 1


@dataclass
class PriceTable:
    pro_output: float = 6.0
    pro_input_cache_hit: float = 0.025
    pro_input_cache_miss: float = 3.0
    flash_output: float = 2.0
    flash_input_cache_hit: float = 0.02
    flash_input_cache_miss: float = 1.0


class PeakValleyManager:
    """Peak-valley pricing manager with optimal scheduling"""
    
    def __init__(self):
        self.base = PriceTable()
        self.peak_factor = 2.0
        self.peak_hours = set(range(9, 12)) | set(range(14, 18))
    
    def get_tier(self, dt: datetime = None) -> Tier:
        dt = dt or datetime.now()
        return Tier.PEAK if dt.hour in self.peak_hours else Tier.OFF_PEAK
    
    def get_price(self, model: str, output: bool, 
                  cache_hit: bool = False, dt: datetime = None) -> float:
        tier = self.get_tier(dt)
        
        if model == "pro":
            if output:
                price = self.base.pro_output
            elif cache_hit:
                price = self.base.pro_input_cache_hit
            else:
                price = self.base.pro_input_cache_miss
        else:  # flash
            if output:
                price = self.base.flash_output
            elif cache_hit:
                price = self.base.flash_input_cache_hit
            else:
                price = self.base.flash_input_cache_miss
        
        if tier == Tier.PEAK:
            price *= self.peak_factor
        
        return price
    
    def find_optimal_window(self, max_delay_hours: int = 24) -> tuple:
        """Find the cheapest time window within max_delay_hours"""
        now = datetime.now()
        best_cost = float('inf')
        best_time = now
        
        for h in range(max_delay_hours * 4):  # 15-min intervals
            t = now + timedelta(minutes=15 * h)
            cost = self.get_price("flash", True, dt=t)
            if cost < best_cost:
                best_cost = cost
                best_time = t
        
        current_cost = self.get_price("flash", True)
        saving = (current_cost - best_cost) / current_cost * 100
        
        return best_time, saving
    
    def cost_estimate(self, 
                      output_tokens: int,
                      input_tokens: int,
                      model: str = "flash",
                      cache_hit_rate: float = 0.3,
                      peak_ratio: float = 0.5) -> dict:
        """Estimate total cost with scheduling optimization"""
        
        output_m = output_tokens / 1_000_000
        input_m = input_tokens / 1_000_000
        
        # Baseline: no optimization
        if model == "pro":
            base_cost = output_m * self.base.pro_output + \
                        input_m * self.base.pro_input_cache_miss
        else:
            base_cost = output_m * self.base.flash_output + \
                        input_m * self.base.flash_input_cache_miss
        
        # With cache optimization
        cache_cost = output_m * (
            self.get_price(model, True, True) * cache_hit_rate +
            self.get_price(model, True, False) * (1 - cache_hit_rate)
        ) + input_m * (
            self.get_price(model, False, True) * cache_hit_rate +
            self.get_price(model, False, False) * (1 - cache_hit_rate)
        )
        
        # With peak-valley optimization
        off_peak_cost = self.get_price(model, True, dt=datetime(2026, 7, 1, 3, 0))
        peak_cost = self.get_price(model, True, dt=datetime(2026, 7, 1, 10, 0))
        optimized_cost = output_m * (
            off_peak_cost * (1 - peak_ratio) + peak_cost * peak_ratio
        ) + cache_cost * 0.5  # simplified
        
        return {
            "base_cost": base_cost,
            "cache_optimized": cache_cost,
            "fully_optimized": optimized_cost,
            "saving_vs_base": (base_cost - optimized_cost) / base_cost * 100,
        }

V. Competitive Cost Efficiency

Dimension DeepSeek V4 Pro DeepSeek V4 Flash GPT-4o Claude Opus 4.8
Output price (M tokens) 6 RMB (peak 12) 2 RMB (peak 4) ~105 RMB ~90 RMB
Context length 1M tokens 1M tokens 128K 200K
Speed boost 78% (DSpark) 85% (DSpark) - -
License MIT ✅ MIT ✅ Closed Closed
Daily cost (1M output) 6 RMB 2 RMB 105 RMB 90 RMB
Optimized daily cost 3 RMB 1 RMB 105 RMB 90 RMB

Even at peak double pricing, DeepSeek V4 remains 1/9th the cost of GPT-4o and 1/7th of Claude Opus 4.8. With cache hits and off-peak scheduling, costs become negligible.


VI. Industry Impact

Peak-Valley Pricing Will Become Industry Standard

DeepSeek’s peak-valley pricing marks AI cloud services evolving from “pay-per-use” to “pay-per-time.” This mirrors electricity tiered pricing — time-dimension pricing enables more efficient supply-demand matching.

Within 12 months, other Chinese models (Qwen, GLM, MiniMax) are expected to follow with similar strategies. This is the inevitable result of compute commoditization.

DSpark’s Inference Revolution

DSpark speculative decoding provides a methodology that “improves inference speed without modifying model architecture.” Flash’s 85% speed boost means the same hardware can serve nearly 2x concurrent requests — critical for China’s compute-constrained AI ecosystem.

Domestic Hardware Strategy

DeepSeek V4 deeply integrates with Huawei’s Ascend ecosystem. When Ascend 950 super-nodes ship in bulk in H2 2026, Pro pricing has further room to decrease. Chinese AI is transitioning from “follower” to “price-setter.”


VII. Conclusion

The DeepSeek V4 official release embodies a dual revolution:

Technical: DSA sparse attention pushes long-context inference costs to the floor, DSpark boosts speed by 85%, and MoE architecture + million-token context + MIT license creates powerful developer pull.

Commercial: Peak-valley pricing isn’t a simple price hike — it’s the landmark event in AI cloud fine-grained operations. It optimizes resource allocation through price leverage while setting the “time-dimension pricing” benchmark for the industry.

As DeepSeek founder Liang Wenfeng stated after the Series A funding: “Technical capability sets the ceiling; the business model sets the floor.” V4 is breaking through both simultaneously.


Sources: DeepSeek official announcement (2026-06-29), IT Home, Southern Metropolis Daily, Sina Finance