Tableau AI增强数据可视化分析

from dataclasses import dataclassfrom typing import Dict, List, Optionalfrom collections import defaultdictimport reimport random @dataclassclass DataField:    name: str    field_type: str    data_type: str    description: str    values: Optional[List] = None @dataclassclass SemanticModel:    fields: Dict[str, DataField]    term_mappings: Dict[str, str]        def resolve_term(self, term: str) -> Optional[str]:        term = term.lower().strip()        if term in self.term_mappings:            return self.term_mappings[term]        for t, f in self.term_mappings.items():            if term in t or t in term:                return f        return None class NLQParser:    def __init__(self, semantic_model: SemanticModel):        self.semantic_model = semantic_model        self.time_patterns = [(r'(今年|本年)', 'current_year'), (r'(去年|上年)', 'last_year'),                              (r'(本月|当月)', 'current_month'), (r'(上月)', 'last_month'),                              (r'(Q[1-4])', 'quarter')]        self.agg_patterns = [(r'(总计|总和|合计)', 'sum'), (r'(平均|均值)', 'avg'),                             (r'(数量|个数)', 'count'), (r'(环比)', 'mom'), (r'(同比)', 'yoy')]        def parse(self, question: str) -> Dict:        result = {'original': question, 'measures': [], 'dimensions': [],                   'filters': [], 'time_range': None, 'aggregation': None, 'chart_type': None}        for term, field_name in self.semantic_model.term_mappings.items():            if term in question:                field = self.semantic_model.fields[field_name]                if field.field_type == 'measure':                    result['measures'].append(field_name)                elif field.field_type == 'dimension':                    result['dimensions'].append(field_name)        for pattern, time_type in self.time_patterns:            if re.search(pattern, question):                result['time_range'] = time_type                break        for pattern, agg_type in self.agg_patterns:            if re.search(pattern, question):                result['aggregation'] = agg_type                break        result['chart_type'] = self._infer_chart_type(result)        return result        def _infer_chart_type(self, parsed: Dict) -> str:        has_time = parsed['time_range'] is not None        num_dims = len(parsed['dimensions'])        num_meas = len(parsed['measures'])        if has_time and num_meas >= 1:            return 'line_chart'        elif num_dims >= 1 and num_meas >= 1:            return 'bar_chart'        else:            return 'table' class ChartRecommender:    @staticmethod    def recommend(data_profile: Dict) -> List[Dict]:        recs = []        dims = data_profile.get('dimensions', [])        meas = data_profile.get('measures', [])        num_dims = len(dims)        num_meas = len(meas)        has_time = any(d.get('data_type') == 'date' for d in dims)        high_card = any(d.get('cardinality', 0) > 20 for d in dims)        if has_time and num_meas >= 1:            recs.append({'chart_type': 'line_chart', 'confidence': 0.9, 'reason': '时间+度量适合趋势展示'})        if num_dims >= 1 and num_meas >= 1 and not high_card:            recs.append({'chart_type': 'bar_chart', 'confidence': 0.8, 'reason': '类别对比柱状图直观'})        if num_dims == 1 and num_meas == 1 and not high_card:            recs.append({'chart_type': 'pie_chart', 'confidence': 0.6, 'reason': '单维度可展示占比'})        if high_card and num_meas >= 1:            recs.append({'chart_type': 'horizontal_bar', 'confidence': 0.75, 'reason': '高基数适合横向条形图'})        return sorted(recs, key=lambda x: x['confidence'], reverse=True) class InsightAnalyzer:    def __init__(self, sig_threshold=0.05, min_effect=0.01):        self.sig_threshold = sig_threshold        self.min_effect = min_effect        def analyze(self, data: List[Dict], dimensions: List[str], measure: str) -> List[Dict]:        insights = []        for dim in dimensions:            groups = defaultdict(list)            for row in data:                groups[row[dim]].append(row[measure])            if len(groups) < 2:                continue            group_stats = {g: {'mean': sum(v)/len(v), 'count': len(v)} for g, v in groups.items()}            sorted_groups = sorted(group_stats.items(), key=lambda x: x[1]['mean'], reverse=True)            top_g = sorted_groups[0]            bottom_g = sorted_groups[-1]            overall_mean = sum(r[measure] for r in data) / len(data)            effect = (top_g[1]['mean'] - bottom_g[1]['mean']) / overall_mean if overall_mean else 0            p_value = max(0.001, 0.1 - abs(effect) * 0.5)            sig = p_value < self.sig_threshold            meaningful = abs(effect) >= self.min_effect            if sig and meaningful:                contribution = (top_g[1]['mean'] - bottom_g[1]['mean']) / overall_mean if overall_mean else 0                insights.append({                    'dimension': dim, 'top_group': top_g[0], 'bottom_group': bottom_g[0],                    'effect_size': round(effect, 4), 'contribution': round(contribution, 4),                    'p_value': round(p_value, 4), 'is_causal': False,                    'description': f'按{dim}分组发现，{top_g[0]}的{measure}显著高于{bottom_g[0]}',                    'severity': 'high' if abs(effect) > 0.1 else 'medium'                })        return sorted(insights, key=lambda x: abs(x['contribution']), reverse=True) def demo():    print('=== Tableau AI 数据可视化模拟器 ===')    fields = {        'sales': DataField('sales', 'measure', 'number', '销售额'),        'profit': DataField('profit', 'measure', 'number', '利润'),        'region': DataField('region', 'dimension', 'string', '地区', ['华东','华南','华北']),        'category': DataField('category', 'dimension', 'string', '品类', ['美妆','食品','服饰']),        'date': DataField('date', 'dimension', 'date', '日期'),    }    term_map = {'销售额': 'sales', '销量': 'sales', '利润': 'profit',                '地区': 'region', '品类': 'category', '日期': 'date'}    sm = SemanticModel(fields, term_map)    parser = NLQParser(sm)    print('\n场景1: NLQ解析')    for q in ['华东区今年销售额', '各品类利润对比', 'Top10城市销量排名']:        r = parser.parse(q)        print(f'  {q} -> 度量:{r["measures"]} 维度:{r["dimensions"]} 图表:{r["chart_type"]}')    print('\n场景2: 自动图表推荐')    prof = {'dimensions': [{'name': 'date', 'data_type': 'date', 'cardinality': 365}], 'measures': ['sales']}    recs = ChartRecommender.recommend(prof)    for r in recs[:2]:        print(f'  - {r["chart_type"]} (置信度:{r["confidence"]}): {r["reason"]}')    print('\n场景3: 智能洞察分析')    random.seed(42)    data = []    regions = ['华东', '华南', '华北', '西南', '华中']    categories = ['美妆', '食品', '服饰', '家电']    for _ in range(500):        region = random.choice(regions)        cat = random.choice(categories)        base = 100        if region == '华中': base *= 0.6        if cat == '美妆': base *= 0.7        data.append({'region': region, 'category': cat, 'sales': max(10, base + random.gauss(0, 20))})    analyzer = InsightAnalyzer(sig_threshold=0.05, min_effect=0.05)    insights = analyzer.analyze(data, ['region', 'category'], 'sales')    print(f'  发现{len(insights)}条显著洞察')    for ins in insights[:2]:        print(f'  [{ins["severity"]}] {ins["description"]}')        print(f'    效应量:{ins["effect_size"]*100:.1f}% p值:{ins["p_value"]:.4f} 因果:{ins["is_causal"]}') if __name__ == '__main__':    demo()