# qviz

**Repository Path**: XimoKB/qviz

## Basic Information

- **Project Name**: qviz
- **Description**: 这是一个基于Streamlit开发的数据挖掘与可视化平台，集成了多种数据处理、分析和可视化功能，支持多源数据整合、高级图表展示、数据模式识别和智能指标监控，帮助用户从复杂数据中获取洞察和辅助决策。平台还包含了专业领域的数据分析能力，如特定数据集的模式识别和策略分析。
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2025-05-03
- **Last Updated**: 2026-02-24

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# BEIZHI - 智能数据分析平台 🚀

![Python](https://img.shields.io/badge/Python-3.8%2B-blue)
![Streamlit](https://img.shields.io/badge/Streamlit-1.24%2B-red)
![License](https://img.shields.io/badge/License-MIT-green)

## 项目概述 📋

BEIZHI是一个基于Streamlit开发的企业级智能数据分析平台，专为金融投资、量化研究和数据科学团队设计。平台集成了数据维护、投资管理、量化研究和智能分析等核心功能模块，支持多种数据源接入、实时数据处理、高级可视化和智能决策支持。

### 核心特性 ✨

- 🔐 **企业级认证系统** - 基于Token的安全认证机制，支持会话管理
- 📊 **数据维护工具** - 现货数据维护、黑色系现货更新等数据管理功能
- 📈 **投资管理模块** - 基差调仓、东财反向分析等投资决策工具
- 🔬 **量化研究平台** - 滚动收益率分析、数据完整性检查、策略业绩跟踪
- 🤖 **AI智能模块** - 财经助手、数据爬虫等智能化工具
- 🏗️ **模块化架构** - 清晰的分层架构，支持功能扩展和维护
- 🎨 **现代化UI** - 响应式设计，优秀的用户体验

## 技术架构 🏗️

### 系统架构图
```
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Web Frontend │    │  Streamlit App  │    │   Data Sources  │
│   (Browser)    │◄──►│    (Python)     │◄──►│  (DB/API/File)  │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                              │
                       ┌─────────────────┐
                       │   Services      │
                       │  (Auth/Data)    │
                       └─────────────────┘
```

### 目录结构
```
qviz/
├── 🚪 app.py                     # 应用入口点
├── 🏃 run_app.py                 # 生产环境启动脚本
├── 📦 requirements.txt           # Python依赖
├── 📄 README.md                  # 项目文档
├── 🗂️ modules/                   # 功能模块
│   ├── 🏠 home.py                # 首页模块
│   ├── 📊 basis_adjust.py        # 基差调仓
│   ├── 📈 dfcf_reverse.py        # 东财反向分析
│   ├── 📋 simple_table_parser.py # 现货数据维护
│   ├── 📊 actuals_update.py      # 黑色系现货更新
│   ├── 🔬 quant_research/        # 量化研究模块
│   │   ├── rolling_returns.py   # 滚动收益率分析
│   │   ├── data_integrity.py    # 数据完整性检查
│   │   ├── main_sub_spread.py   # 主力次主力价差
│   │   └── strategy_performance.py # 策略业绩跟踪
│   ├── 🤖 finance_assistant/     # 财经助手
│   └── 🕸️ finance_crawler/      # 财经数据爬虫
├── 🔧 services/                  # 业务服务层
│   ├── 🔐 auth.py                # 认证服务
│   ├── 🗄️ database.py           # 数据库服务
│   ├── 🛠️ database_utils.py     # 数据库工具
│   ├── 📊 table_extraction.py    # 表格提取服务
│   └── 👁️ blink_detection.py    # 变化检测服务
├── 🎨 components/               # UI组件
├── 🛠️ utils/                    # 工具函数
├── 📁 models/                   # 数据模型
├── ⚙️ config/                   # 配置文件
├── 🧪 tests/                    # 测试文件
└── 🎛️ .streamlit/              # Streamlit配置
    ├── config.toml
    └── pages.toml
```

## 快速开始 🚀

### 环境要求
- Python 3.8+
- 8GB+ RAM推荐
- 支持HTTPS的Web服务器（生产环境）

### 本地开发部署

1. **克隆项目**
   ```bash
   git clone <repository-url>
   cd qviz
   ```

2. **创建虚拟环境**
   ```bash
   python -m venv venv
   source venv/bin/activate  # Linux/Mac
   # 或 venv\Scripts\activate  # Windows
   ```

3. **安装依赖**
   ```bash
   pip install -r requirements.txt
   ```

4. **环境配置**
   
   创建 `.env` 文件：
   ```bash
   cp .env.example .env
   ```
   
   配置必要的环境变量：
   ```env
   # 基础认证配置
   AUTH_USERNAME=admin
   AUTH_PASSWORD=your_secure_password
   AUTH_SECRET_KEY=your_secret_key_here
   AUTH_SESSION_EXPIRY=86400
   
   # 数据库配置
   PG_DB_NAME=your_pg_db_name
   PG_DB_USER=your_username
   PG_DB_PASSWORD=your_password
   PG_DB_HOST=localhost
   PG_DB_PORT=5432
   
   # API Keys
   DASHSCOPE_API_KEY=your_dashscope_api_key
   DATAYES_API_KEY=your_datayes_api_key
   OPENAI_API_KEY=your_openai_api_key
   ```

5. **启动应用**
   ```bash
   # 开发模式
   streamlit run app.py
   
   # 或使用启动脚本
   python run_app.py
   ```

6. **访问应用**
   
   打开浏览器访问: http://localhost:8501

## 生产环境部署 🌐

### 推荐部署架构

```
Internet → Load Balancer → Reverse Proxy (Nginx) → Streamlit App → Database
```

### Docker容器化部署

1. **创建Dockerfile**
   ```dockerfile
   FROM python:3.9-slim
   
   WORKDIR /app
   COPY requirements.txt .
   RUN pip install --no-cache-dir -r requirements.txt
   
   COPY . .
   
   EXPOSE 8501
   
   HEALTHCHECK CMD curl --fail http://localhost:8501/_stcore/health
   
   CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]
   ```

2. **构建和运行**
   ```bash
   docker build -t beizhi-platform .
   docker run -p 8501:8501 --env-file .env beizhi-platform
   ```

### Nginx反向代理配置

```nginx
server {
    listen 80;
    server_name your-domain.com;
    
    location / {
        proxy_pass http://localhost:8501;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # WebSocket支持
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}
```

## 生产环境优化建议 🔧

### 1. 性能优化

#### 当前问题：
- ❌ 缺少数据缓存机制
- ❌ 没有使用连接池
- ❌ 缺少异步数据处理
- ❌ 静态资源未优化

#### 优化方案：

**1.1 数据缓存优化**
```python
# 在相关模块中添加缓存装饰器
@st.cache_data(ttl=3600, max_entries=100)
def load_financial_data(symbol, start_date, end_date):
    # 数据加载逻辑
    pass

@st.cache_resource
def get_database_connection():
    # 数据库连接逻辑
    pass
```

**1.2 数据库连接池**
```python
# 在 services/database.py 中实现
from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20,
    pool_recycle=3600
)
```

**1.3 异步数据处理**
```python
# 添加后台任务处理
import asyncio
from concurrent.futures import ThreadPoolExecutor

@st.cache_data
def process_data_async(data):
    with ThreadPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(process_chunk, data_chunks))
    return results
```

### 2. 安全性增强

#### 当前问题：
- ❌ 简单的Token认证机制
- ❌ 缺少输入验证
- ❌ 没有API访问限制
- ❌ 敏感数据未加密

#### 优化方案：

**2.1 增强认证系统**
```python
# 升级为JWT Token
import jwt
from datetime import datetime, timedelta

def create_jwt_token(username):
    payload = {
        'username': username,
        'exp': datetime.utcnow() + timedelta(hours=24),
        'iat': datetime.utcnow()
    }
    return jwt.encode(payload, SECRET_KEY, algorithm='HS256')
```

**2.2 输入验证和清理**
```python
# 添加输入验证装饰器
from functools import wraps
import re

def validate_input(validation_rules):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 验证逻辑
            return func(*args, **kwargs)
        return wrapper
    return decorator
```

**2.3 API访问限制**
```python
# 添加速率限制
from functools import wraps
import time

def rate_limit(max_calls=100, time_window=3600):
    def decorator(func):
        # 实现速率限制逻辑
        pass
    return decorator
```

### 3. 监控和日志

#### 需要添加的组件：

**3.1 应用监控**
```python
# 添加 monitoring.py
import logging
import psutil
import streamlit as st

def log_performance_metrics():
    cpu_usage = psutil.cpu_percent()
    memory_usage = psutil.virtual_memory().percent
    
    logging.info(f"CPU: {cpu_usage}%, Memory: {memory_usage}%")
```

**3.2 错误追踪**
```python
# 添加错误处理中间件
import traceback
import logging

def error_handler(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logging.error(f"Error in {func.__name__}: {str(e)}")
            logging.error(traceback.format_exc())
            st.error("系统错误，请联系管理员")
    return wrapper
```

### 4. 代码质量提升

#### 需要实施的改进：

**4.1 添加类型提示**
```python
from typing import Dict, List, Optional, Union
import pandas as pd

def process_financial_data(
    data: pd.DataFrame,
    columns: List[str],
    config: Dict[str, Union[str, int]]
) -> Optional[pd.DataFrame]:
    pass
```

**4.2 单元测试覆盖**
```python
# tests/test_auth.py
import pytest
from services.auth import create_token, verify_token

def test_token_creation():
    token = create_token("test_user")
    assert token is not None
    
def test_token_verification():
    token = create_token("test_user")
    is_valid, username = verify_token(token)
    assert is_valid is True
    assert username == "test_user"
```

**4.3 配置管理优化**
```python
# config/settings.py
from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "BEIZHI"
    debug: bool = False
    database_url: str
    secret_key: str
    
    class Config:
        env_file = ".env"

settings = Settings()
```

### 5. 部署和DevOps

#### 推荐的CI/CD流程：

**5.1 GitHub Actions配置**
```yaml
# .github/workflows/ci.yml
name: CI/CD Pipeline

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: 3.9
    - name: Install dependencies
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-cov black flake8
    - name: Run tests
      run: |
        black --check .
        flake8 .
        pytest --cov=.
```

**5.2 Docker Compose部署**
```yaml
# docker-compose.yml
version: '3.8'
services:
  app:
    build: .
    ports:
      - "8501:8501"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/beizhi
    depends_on:
      - db
      - redis
  
  db:
    image: postgres:13
    environment:
      POSTGRES_DB: beizhi
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  redis:
    image: redis:alpine
    
volumes:
  postgres_data:
```

## 功能模块详解 📚

### 数据维护模块
- **现货数据维护**: 支持多种格式的现货数据导入、清洗和更新
- **黑色系现货更新**: 专门针对黑色金属现货数据的批量更新工具

### 投资管理模块
- **基差调仓**: 期货基差分析和调仓建议系统
- **东财反向分析**: 基于东方财富数据的反向投资策略分析

### 量化研究模块
- **滚动收益率分析**: 多时间窗口的收益率统计和可视化
- **数据完整性检查**: 数据质量评估和缺失值分析
- **主力次主力价差**: 期货合约价差分析和套利机会识别
- **策略业绩跟踪**: 投资策略回测和业绩监控

### AI智能模块
- **财经助手**: 基于LLM的财经数据问答和分析
- **数据爬虫**: 自动化财经网站数据采集工具

## 开发指南 👨‍💻

### 添加新功能模块

1. **创建模块文件**
   ```python
   # modules/new_feature.py
   import streamlit as st
   
   # 禁用自动页面发现
   __title__ = None
   __is_not_streamlit_page__ = True
   
   def show_new_feature():
       st.title("新功能")
       # 功能实现
   ```

2. **注册路由**
   ```python
   # 在 app.py 中添加
   from modules.new_feature import show_new_feature
   
   # 在 display_current_page() 中添加路由
   elif st.session_state.current_page == "new_feature":
       show_new_feature()
   ```

### 代码规范

- 使用 `black` 进行代码格式化
- 使用 `flake8` 进行代码检查
- 添加类型提示和文档字符串
- 遵循 PEP 8 编码规范

### 测试策略

```bash
# 运行测试
pytest tests/ -v

# 生成覆盖率报告
pytest --cov=modules --cov=services tests/

# 生成HTML覆盖率报告
pytest --cov=modules --cov=services --cov-report=html tests/
```

## 常见问题 ❓

### Q: 如何添加新的数据源？
A: 在 `services/` 目录下创建新的服务文件，实现数据连接和处理逻辑，然后在相应的页面模块中调用。

### Q: 如何优化大数据集的处理性能？
A: 使用 `@st.cache_data` 缓存数据，实现数据分块处理，使用数据库连接池。

### Q: 如何部署到云服务器？
A: 参考上面的Docker部署方案，或使用Streamlit Cloud进行快速部署。

## 贡献指南 🤝

1. Fork 项目
2. 创建功能分支 (`git checkout -b feature/AmazingFeature`)
3. 提交更改 (`git commit -m 'Add some AmazingFeature'`)
4. 推送到分支 (`git push origin feature/AmazingFeature`)
5. 打开 Pull Request

## 版本历史 📋

- **v2.0.0** - 企业级重构，添加认证系统和模块化架构
- **v1.5.0** - 添加量化研究模块和AI助手
- **v1.0.0** - 基础数据分析功能实现

## 许可证 📄

本项目采用 MIT 许可证 - 查看 [LICENSE](LICENSE) 文件了解详情

## 联系方式 📧

- 项目维护者: [您的姓名](mailto:your.email@example.com)
- 项目链接: [https://github.com/yourname/qviz](https://github.com/yourname/qviz)

---

⭐ 如果这个项目对您有帮助，请给它一个星标！