market-data-service/internal/monitor/monitor.go

package monitor

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

	"market-data-service/api"
	"market-data-service/internal/repository"
)

// Monitor 数据质量监控
type Monitor struct {
	db          *repository.DB
	stockRepo   *repository.StockRepository
	futuresRepo *repository.FuturesRepository
	sender      AlertSender
}

// AlertSender 告警发送接口
type AlertSender interface {
	SendAlert(title, content string) error
}

// NewMonitor 创建监控器
func NewMonitor(db *repository.DB, stockRepo *repository.StockRepository, futuresRepo *repository.FuturesRepository, sender AlertSender) *Monitor {
	return &Monitor{
		db:          db,
		stockRepo:   stockRepo,
		futuresRepo: futuresRepo,
		sender:      sender,
	}
}

// CheckResult 检查结果
type CheckResult struct {
	Symbol      string
	Freq        string
	CheckDate   string
	CheckType   string
	Status      string // pass/fail
	ExpectCount int
	ActualCount int
	Detail      string
}

// DailyCheck 每日数据质量检查
func (m *Monitor) DailyCheck(ctx context.Context, checkDate string) error {
	log.Printf("Starting daily data quality check for %s", checkDate)

	// 检查股票数据
	if err := m.checkStockData(ctx, checkDate); err != nil {
		log.Printf("Stock data check failed: %v", err)
	}

	// 检查期货数据
	if err := m.checkFuturesData(ctx, checkDate); err != nil {
		log.Printf("Futures data check failed: %v", err)
	}

	log.Printf("Daily data quality check completed")
	return nil
}

// checkStockData 检查股票数据质量
func (m *Monitor) checkStockData(ctx context.Context, checkDate string) error {
	// 1. 获取所有活跃股票
	symbols, _, err := m.stockRepo.ListSymbols(ctx, &api.SymbolListRequest{
		Page: 1,
		Size: 10000,
	})
	if err != nil {
		return err
	}

	// 2. 检查1分钟线完整性（240条）
	for _, symbol := range symbols {
		result := m.checkKLineCompleteness(ctx, "stock", symbol.SymbolID, "1m", checkDate, 240)
		if err := m.saveCheckResult(ctx, "stock", result); err != nil {
			log.Printf("Failed to save check result: %v", err)
		}
	}

	return nil
}

// checkFuturesData 检查期货数据质量
func (m *Monitor) checkFuturesData(ctx context.Context, checkDate string) error {
	// 获取所有活跃期货合约
	symbols, _, err := m.futuresRepo.ListSymbols(ctx, &api.SymbolListRequest{
		Page: 1,
		Size: 10000,
	})
	if err != nil {
		return err
	}

	// 检查1分钟线完整性
	for _, symbol := range symbols {
		result := m.checkKLineCompleteness(ctx, "futures", symbol.SymbolID, "1m", checkDate, 240)
		if err := m.saveCheckResult(ctx, "futures", result); err != nil {
			log.Printf("Failed to save check result: %v", err)
		}
	}

	return nil
}

// checkKLineCompleteness 检查K线完整性
func (m *Monitor) checkKLineCompleteness(ctx context.Context, assetType, symbol, freq, checkDate string, expectCount int) CheckResult {
	result := CheckResult{
		Symbol:    symbol,
		Freq:      freq,
		CheckDate: checkDate,
		CheckType: "missing",
		Status:    "pass",
	}

	// 解析日期
	start, _ := time.Parse("20060102", checkDate)
	end := start.Add(24 * time.Hour).Add(-time.Second)

	var actualCount int
	var err error

	if assetType == "stock" {
		items, err := m.stockRepo.GetKLines(ctx, symbol, api.Frequency(freq), start, end, api.AdjustNone)
		if err == nil {
			actualCount = len(items)
		}
	} else {
		items, err := m.futuresRepo.GetKLines(ctx, symbol, api.Frequency(freq), start, end)
		if err == nil {
			actualCount = len(items)
		}
	}

	if err != nil {
		result.Status = "fail"
		result.Detail = fmt.Sprintf("Error querying data: %v", err)
		return result
	}

	result.ExpectCount = expectCount
	result.ActualCount = actualCount

	// 判断缺失情况
	if actualCount < expectCount {
		result.Status = "fail"
		result.Detail = fmt.Sprintf("Data missing: expected %d, actual %d", expectCount, actualCount)
		
		// 发送告警
		if m.sender != nil {
			title := fmt.Sprintf("[%s] Data Missing Alert", assetType)
			content := fmt.Sprintf("Symbol: %s, Date: %s, Expected: %d, Actual: %d",
				symbol, checkDate, expectCount, actualCount)
			m.sender.SendAlert(title, content)
		}
	}

	return result
}

// saveCheckResult 保存检查结果
func (m *Monitor) saveCheckResult(ctx context.Context, assetType string, result CheckResult) error {
	// 使用Repository保存到data_quality_checks表
	query := fmt.Sprintf(`
		INSERT INTO %s.data_quality_checks 
		(check_date, symbol_id, freq, check_type, status, expect_count, actual_count, detail)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
		ON CONFLICT (check_date, symbol_id, freq, check_type) DO UPDATE SET
			status = EXCLUDED.status,
			expect_count = EXCLUDED.expect_count,
			actual_count = EXCLUDED.actual_count,
			detail = EXCLUDED.detail,
			created_at = NOW()
	`, assetType)

	_, err := m.db.ExecContext(ctx, query,
		result.CheckDate, result.Symbol, result.Freq,
		result.CheckType, result.Status, result.ExpectCount,
		result.ActualCount, result.Detail)

	return err
}

// GetQualityReport 获取数据质量报告
func (m *Monitor) GetQualityReport(ctx context.Context, assetType, checkDate string) (*QualityReport, error) {
	query := fmt.Sprintf(`
		SELECT 
			COUNT(*) as total_checks,
			SUM(CASE WHEN status = 'pass' THEN 1 ELSE 0 END) as pass_count,
			SUM(CASE WHEN status = 'fail' THEN 1 ELSE 0 END) as fail_count
		FROM %s.data_quality_checks
		WHERE check_date = $1
	`, assetType)

	var report QualityReport
	report.CheckDate = checkDate
	report.AssetType = assetType

	row := m.db.QueryRowContext(ctx, query, checkDate)
	err := row.Scan(&report.TotalChecks, &report.PassCount, &report.FailCount)
	if err != nil {
		return nil, err
	}

	if report.TotalChecks > 0 {
		report.PassRate = float64(report.PassCount) / float64(report.TotalChecks) * 100
	}

	return &report, nil
}

// QualityReport 数据质量报告
type QualityReport struct {
	AssetType   string  `json:"asset_type"`
	CheckDate   string  `json:"check_date"`
	TotalChecks int     `json:"total_checks"`
	PassCount   int     `json:"pass_count"`
	FailCount   int     `json:"fail_count"`
	PassRate    float64 `json:"pass_rate"`
}

// LogAlertSender 日志告警发送器
type LogAlertSender struct{}

// SendAlert 发送告警到日志
func (s *LogAlertSender) SendAlert(title, content string) error {
	log.Printf("[ALERT] %s: %s", title, content)
	return nil
}

// StartDailyCheckCron 启动每日检查定时任务
func (m *Monitor) StartDailyCheckCron(ctx context.Context) {
	go func() {
		for {
			// 计算到下一个盘后的时间（假设15:30）
			now := time.Now()
			next := time.Date(now.Year(), now.Month(), now.Day(), 15, 35, 0, 0, now.Location())
			if next.Before(now) {
				next = next.Add(24 * time.Hour)
			}

			timer := time.NewTimer(next.Sub(now))
			<-timer.C

			// 执行检查
			checkDate := now.Format("20060102")
			if err := m.DailyCheck(ctx, checkDate); err != nil {
				log.Printf("Daily check failed: %v", err)
			}
		}
	}()
}