stock-analysis/internal/datasource/fallback/fallback.go

// Package fallback 实现 datasource.MarketDataSource 接口，提供数据源主备切换能力。
//
// 主数据源（如东财）用于实时行情和历史K线；当主数据源历史K线请求失败时，
// 自动切换到备用数据源（如腾讯），确保数据入库和回测的可用性。
// 实时行情（FetchConvertibles）支持主备切换：主数据源失败时，
// 走"东财 datacenter 列表 + 腾讯实时报价"的组合策略补充实时价格。
//
// 为提高数据源故障时的整体效率，fallback 内置熔断机制：主数据源K线连续
// 失败达到阈值后，进入熔断窗口，期间直接使用备用数据源，不再尝试主数据源；
// 熔断窗口结束后自动放行一次探测，成功则解除熔断，恢复主备切换。
package fallback

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	"stock-analysis/internal/datasource"
	"stock-analysis/internal/datasource/eastmoney"
	"stock-analysis/internal/datasource/tencent"
	"stock-analysis/internal/model"
)

const (
	// circuitThreshold 主数据源连续失败达到该次数后触发熔断。
	circuitThreshold = 3
	// circuitCooldown 熔断持续时间，之后放行一次探测请求。
	circuitCooldown = 10 * time.Minute
)

// New 创建主备切换数据源。
// primary 为主数据源（如东财），secondary 为备用数据源（如腾讯）。
func New(primary, secondary datasource.MarketDataSource) datasource.MarketDataSource {
	return &dataSource{
		primary:   primary,
		secondary: secondary,
	}
}

// dataSource 包装主备两个数据源，实现 MarketDataSource 接口。
type dataSource struct {
	primary   datasource.MarketDataSource
	secondary datasource.MarketDataSource

	mu           sync.Mutex
	failCount    int       // 主数据源连续失败次数
	openSince    time.Time // 熔断开启时间（open=true 时有效）
	open         bool      // 熔断是否开启
}

// FetchConvertibles 拉取全量可转债实时行情：先尝试主数据源 push2 接口，
// 失败时走"东财 datacenter 列表 + 腾讯批量实时报价"的组合策略补充实时价格。
//
// 主数据源 push2 故障是常见场景（东财对该集群限流较严），datacenter 与 push2 分属
// 不同集群，通常仍可用，故组合策略能保证实时行情可用性。
func (d *dataSource) FetchConvertibles(ctx context.Context) ([]model.ConvertibleBond, error) {
	bonds, err := d.primary.FetchConvertibles(ctx)
	if err == nil {
		return bonds, nil
	}

	log.Printf("[fallback] 主数据源实时行情失败: %v，切换到组合策略（东财列表 + 腾讯报价）", err)
	return d.fetchConvertiblesViaCombo(ctx)
}

// fetchConvertiblesViaCombo 组合策略实时行情：
//  1. 东财 datacenter 拉转债列表（含正股代码、初始转股价、上市日期）
//  2. 腾讯批量实时接口（qt.gtimg.cn）补充实时价格和涨跌幅
//  3. 过滤已退市转债（实时价格为 0）
func (d *dataSource) fetchConvertiblesViaCombo(ctx context.Context) ([]model.ConvertibleBond, error) {
	em := eastmoney.NewClient()
	bonds, err := em.FetchAllBondCodes(ctx)
	if err != nil {
		return nil, fmt.Errorf("datacenter 拉取列表失败: %w", err)
	}
	if len(bonds) == 0 {
		return nil, fmt.Errorf("datacenter 返回空列表")
	}
	log.Printf("[fallback] datacenter 拉取到 %d 只转债，开始腾讯实时报价", len(bonds))

	if err := tencent.FetchQuotes(ctx, bonds); err != nil {
		// 腾讯报价失败不致命，已退市转债会被 price=0 过滤
		log.Printf("[fallback] 腾讯实时报价异常: %v", err)
	}

	// 过滤已退市转债（实时价格为 0），与项目硬约束一致
	out := make([]model.ConvertibleBond, 0, len(bonds))
	for _, b := range bonds {
		if b.Price > 0 {
			out = append(out, b)
		}
	}
	log.Printf("[fallback] 组合策略完成：有效转债 %d 只（已过滤退市 %d 只）", len(out), len(bonds)-len(out))
	return out, nil
}

// FetchKline 获取历史K线：先尝试主数据源，失败后自动切换到备用数据源。
// 熔断开启时直接使用备用数据源。
func (d *dataSource) FetchKline(ctx context.Context, code, klt, beg, end string) ([]model.Kline, error) {
	// 熔断开启：直接走备用数据源
	if d.circuitOpen() {
		return d.fetchSecondary(ctx, code, klt, beg, end)
	}

	klines, err := d.primary.FetchKline(ctx, code, klt, beg, end)
	if err == nil {
		// 主数据源恢复，重置失败计数
		d.recordResult(true)
		return klines, nil
	}

	log.Printf("[fallback] 主数据源 K线失败 (%s %s %s~%s): %v，切换到备用数据源", code, klt, beg, end, err)
	d.recordResult(false)
	return d.fetchSecondary(ctx, code, klt, beg, end)
}

// fetchSecondary 从备用数据源获取K线并记录错误。
func (d *dataSource) fetchSecondary(ctx context.Context, code, klt, beg, end string) ([]model.Kline, error) {
	klines, err := d.secondary.FetchKline(ctx, code, klt, beg, end)
	if err != nil {
		return nil, fmt.Errorf("备用数据源: %w", err)
	}
	return klines, nil
}

// circuitOpen 判断是否处于熔断开启状态；熔断超时后自动放行探测。
func (d *dataSource) circuitOpen() bool {
	d.mu.Lock()
	defer d.mu.Unlock()
	if !d.open {
		return false
	}
	if time.Since(d.openSince) >= circuitCooldown {
		// 熔断窗口结束，放行一次探测请求
		d.open = false
		d.failCount = 0
		log.Printf("[fallback] 熔断窗口结束，放行主数据源探测")
		return false
	}
	return true
}

// recordResult 记录主数据源请求结果：失败达到阈值时开启熔断。
func (d *dataSource) recordResult(ok bool) {
	d.mu.Lock()
	defer d.mu.Unlock()
	if ok {
		d.failCount = 0
		if d.open {
			d.open = false
			log.Printf("[fallback] 主数据源恢复，解除熔断")
		}
		return
	}
	d.failCount++
	if d.open {
		return
	}
	if d.failCount >= circuitThreshold {
		d.open = true
		d.openSince = time.Now()
		log.Printf("[fallback] 主数据源连续失败 %d 次，开启熔断 %v，直接使用备用数据源", d.failCount, circuitCooldown)
	}
}