ANIMATEDIFF PRO新手指南：从Python爬虫到电影级动画，完整实战流程
·
ANIMATEDIFF PRO新手指南：从Python爬虫到电影级动画，完整实战流程
最近有个朋友问我，能不能用AI把枯燥的数据报表变成动态视频，比如把股票走势、天气变化这些数字，直接生成一段电影质感的动画。我第一反应是：这想法挺酷，但实现起来会不会很复杂？
试了几个方案后，我发现了一个绝佳组合——Python爬虫抓数据，ANIMATEDIFF PRO做动画。你可能听说过AnimateDiff，但ANIMATEDIFF PRO是它的升级版，专门为追求电影级效果而生。今天我就带你从零开始，完整走一遍这个流程。
1. 项目思路：当数据遇见电影级动画
传统的数据可视化是什么样？折线图、柱状图、热力图，这些静态图表虽然能传达信息，但总觉得少了点生命力。如果能让数据“动”起来，而且动得有电影质感，那会是怎样的体验？
我的想法很简单：用Python爬虫实时抓取数据，把这些数字转换成动画参数，然后用ANIMATEDIFF PRO生成电影级的动态视频。听起来有点抽象？我给你看几个实际场景：
场景一：天气变化可视化 早上6点，城市在晨雾中苏醒，色调偏冷，画面柔和；中午12点，阳光明媚，色彩饱和度高，影子清晰；傍晚6点，夕阳西下，暖色调，长影子。这不是天气预报，这是用真实天气数据生成的24小时城市景观动画。
场景二：股票波动可视化 股价上涨时，镜头缓缓上移，画面充满绿色粒子，节奏轻快；股价下跌时，镜头下沉，红色粒子密集，氛围紧张；成交量放大时，粒子数量激增，形成视觉冲击。这不是K线图，这是用实时交易数据生成的金融艺术动画。
场景三：社交媒体情绪可视化 话题热度上升，画面中心发光，粒子向外扩散；负面情绪聚集，色调变暗，粒子向内收缩；讨论激烈时，画面充满动态线条。这不是词云图，这是用舆情数据生成的情感波动动画。
这套方案的核心价值在于：数据驱动创意，AI实现视觉化。你不用懂3D建模，不用学视频剪辑，只要会写Python代码，就能把任何时序数据变成电影级动画。
2. 环境准备：10分钟快速搭建
在开始写代码之前，我们需要准备好工作环境。如果你已经在星图GPU平台上部署了ANIMATEDIFF PRO镜像，那大部分环境已经配置好了。如果没有，跟着我一步步来。
2.1 基础环境检查
首先确认你的Python环境，我推荐使用Python 3.8以上版本。打开终端，运行以下命令检查：
# 检查Python版本
import sys
print(f"Python版本: {sys.version}")
# 检查关键库
required_libs = ['requests', 'pandas', 'numpy', 'json', 'datetime']
missing_libs = []
for lib in required_libs:
try:
__import__(lib)
print(f"✓ {lib} 已安装")
except ImportError:
missing_libs.append(lib)
print(f"✗ {lib} 未安装")
if missing_libs:
print(f"\n需要安装的库: {missing_libs}")
print("运行: pip install " + " ".join(missing_libs))
else:
print("\n所有依赖库已就绪！")
如果发现有库没安装，用pip安装一下：
pip install requests pandas numpy
2.2 ANIMATEDIFF PRO配置确认
ANIMATEDIFF PRO镜像已经预装了所有必要的组件，但我们还是需要确认几个关键点：
-
服务是否启动：在终端运行
bash /root/build/start.sh，看到类似下面的输出就说明启动成功了：Starting AnimateDiff PRO service... Server running on http://localhost:5000 Cinema UI ready! -
访问Web界面：在浏览器打开
http://localhost:5000，你应该能看到一个深色系的专业界面，这就是ANIMATEDIFF PRO的电影级渲染工作台。 -
API接口测试：ANIMATEDIFF PRO提供了REST API，我们可以用Python直接调用。先做个简单的测试：
import requests
def test_animatediff_api():
"""测试ANIMATEDIFF PRO API连接"""
try:
response = requests.get("http://localhost:5000/api/status", timeout=5)
if response.status_code == 200:
print("✓ ANIMATEDIFF PRO API连接正常")
return True
else:
print(f"✗ API返回状态码: {response.status_code}")
return False
except Exception as e:
print(f"✗ 连接失败: {e}")
print("请确认ANIMATEDIFF PRO服务已启动")
return False
# 运行测试
if test_animatediff_api():
print("环境准备完成，可以开始项目了！")
else:
print("请先启动ANIMATEDIFF PRO服务")
2.3 项目目录结构
建议按下面的结构组织你的项目文件，这样代码清晰，管理方便：
data_driven_animation/
├── data/ # 数据存储
│ ├── raw/ # 原始数据
│ ├── processed/ # 处理后的数据
│ └── cache/ # 缓存文件
├── scripts/ # Python脚本
│ ├── crawlers/ # 爬虫脚本
│ ├── mappers/ # 数据映射脚本
│ ├── generators/ # 动画生成脚本
│ └── utils/ # 工具函数
├── outputs/ # 输出文件
│ ├── animations/ # 生成的动画
│ ├── configs/ # 配置文件
│ └── logs/ # 日志文件
├── config.yaml # 项目配置文件
└── main.py # 主程序
创建这个目录结构：
import os
project_structure = {
"data": ["raw", "processed", "cache"],
"scripts": ["crawlers", "mappers", "generators", "utils"],
"outputs": ["animations", "configs", "logs"]
}
for main_dir, sub_dirs in project_structure.items():
os.makedirs(main_dir, exist_ok=True)
for sub_dir in sub_dirs:
os.makedirs(os.path.join(main_dir, sub_dir), exist_ok=True)
print("项目目录结构创建完成！")
3. 数据采集：Python爬虫实战
数据是动画的原料，没有好数据，再好的AI也做不出好动画。这一节我带你写几个实用的爬虫，从简单到复杂，覆盖常见的应用场景。
3.1 天气数据爬虫：从中国天气网获取实时数据
天气数据是最适合做动画可视化的，变化有规律，数据容易获取。我们从一个简单的天气爬虫开始：
import requests
import pandas as pd
from datetime import datetime
import time
import json
class WeatherDataCrawler:
"""天气数据爬虫类"""
def __init__(self, city="北京"):
self.city = city
self.base_url = "http://t.weather.sojson.com/api/weather/city/"
self.data_file = f"data/raw/weather_{city}_{datetime.now().strftime('%Y%m%d')}.json"
# 城市代码映射（简化版，实际需要更完整的映射）
self.city_codes = {
"北京": "101010100",
"上海": "101020100",
"广州": "101280101",
"深圳": "101280601",
"杭州": "101210101",
"成都": "101270101",
"武汉": "101200101",
"西安": "101110101"
}
def get_city_code(self):
"""获取城市代码"""
return self.city_codes.get(self.city, "101010100") # 默认北京
def fetch_current_weather(self):
"""获取当前天气数据"""
city_code = self.get_city_code()
url = f"{self.base_url}{city_code}"
print(f"正在获取 {self.city} 的天气数据...")
try:
# 设置请求头，模拟浏览器访问
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "application/json"
}
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status() # 检查HTTP错误
data = response.json()
if data.get("status") == 200:
weather_info = self._parse_weather_data(data)
print(f"获取成功: {weather_info['temperature']}°C, {weather_info['weather']}")
return weather_info
else:
print(f"API返回错误: {data.get('message')}")
return None
except requests.exceptions.RequestException as e:
print(f"网络请求失败: {e}")
return None
except json.JSONDecodeError as e:
print(f"JSON解析失败: {e}")
return None
except Exception as e:
print(f"未知错误: {e}")
return None
def _parse_weather_data(self, data):
"""解析天气数据"""
weather_info = data.get("data", {})
forecast = weather_info.get("forecast", [{}])[0] # 今天预报
# 构建完整的数据结构
result = {
"timestamp": datetime.now().isoformat(),
"city": self.city,
"city_code": data.get("cityInfo", {}).get("cityId", ""),
# 基础天气信息
"temperature": int(weather_info.get("wendu", 0)), # 温度
"humidity": int(weather_info.get("shidu", "0%").replace("%", "")), # 湿度
"pressure": weather_info.get("pressure", 0), # 气压
# 天气状况
"weather": forecast.get("type", "晴"), # 天气类型
"high_temp": int(forecast.get("high", "0°").replace("高温 ", "").replace("°", "")),
"low_temp": int(forecast.get("low", "0°").replace("低温 ", "").replace("°", "")),
# 风信息
"wind_direction": forecast.get("fx", "无风"), # 风向
"wind_power": forecast.get("fl", "微风"), # 风力
# 其他信息
"sunrise": weather_info.get("sunrise", ""), # 日出
"sunset": weather_info.get("sunset", ""), # 日落
"aqi": data.get("cityInfo", {}).get("aqi", 0), # 空气质量
"quality": data.get("cityInfo", {}).get("quality", ""), # 空气质量等级
# 提示信息
"ganmao": weather_info.get("ganmao", ""), # 感冒提示
"tips": forecast.get("notice", "") # 生活提示
}
return result
def save_weather_data(self, weather_data):
"""保存天气数据"""
if not weather_data:
return False
try:
# 读取现有数据
try:
with open(self.data_file, "r", encoding="utf-8") as f:
existing_data = json.load(f)
if not isinstance(existing_data, list):
existing_data = []
except (FileNotFoundError, json.JSONDecodeError):
existing_data = []
# 添加新数据
existing_data.append(weather_data)
# 保存数据
with open(self.data_file, "w", encoding="utf-8") as f:
json.dump(existing_data, f, ensure_ascii=False, indent=2)
print(f"数据已保存到: {self.data_file}")
return True
except Exception as e:
print(f"保存数据失败: {e}")
return False
def collect_daily_data(self, interval_hours=1, days=1):
"""收集多日数据"""
total_points = days * 24 // interval_hours
collected_data = []
print(f"开始收集 {self.city} 的天气数据")
print(f"采集间隔: {interval_hours}小时，持续天数: {days}天，总采集点: {total_points}")
print("-" * 50)
for i in range(total_points):
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{current_time}] 第 {i+1}/{total_points} 次采集")
# 获取数据
weather_data = self.fetch_current_weather()
if weather_data:
collected_data.append(weather_data)
self.save_weather_data(weather_data)
# 显示最新数据
latest = collected_data[-1]
print(f" 温度: {latest['temperature']}°C | 湿度: {latest['humidity']}%")
print(f" 天气: {latest['weather']} | 风力: {latest['wind_power']}")
# 如果不是最后一次，等待指定时间
if i < total_points - 1:
print(f" 等待 {interval_hours} 小时...")
time.sleep(interval_hours * 3600)
print("-" * 50)
print(f"数据收集完成！共收集 {len(collected_data)} 条记录")
# 转换为DataFrame方便分析
df = pd.DataFrame(collected_data)
csv_file = self.data_file.replace(".json", ".csv")
df.to_csv(csv_file, index=False, encoding="utf-8-sig")
print(f"数据已导出为CSV: {csv_file}")
return df
# 使用示例
if __name__ == "__main__":
# 创建爬虫实例
crawler = WeatherDataCrawler(city="北京")
# 测试单次采集
print("=== 测试单次采集 ===")
test_data = crawler.fetch_current_weather()
if test_data:
print("测试成功！")
print(f"城市: {test_data['city']}")
print(f"温度: {test_data['temperature']}°C")
print(f"天气: {test_data['weather']}")
print(f"湿度: {test_data['humidity']}%")
print(f"风力: {test_data['wind_power']}")
# 开始定时采集（实际使用时取消注释）
# print("\n=== 开始定时采集 ===")
# df = crawler.collect_daily_data(interval_hours=2, days=3)
# print(f"共采集 {len(df)} 条天气数据")
这个爬虫做了几件重要的事：
- 数据完整性：不仅获取温度湿度，还包括日出日落、空气质量、生活提示等丰富信息
- 错误处理：网络异常、JSON解析错误、API返回错误都有相应处理
- 数据持久化：自动保存为JSON和CSV格式，方便后续处理
- 定时采集：支持按小时间隔自动采集，构建时间序列数据
3.2 股票数据爬虫：获取实时金融数据
如果你对金融可视化感兴趣，股票数据是绝佳的动画素材。下面是一个简单的股票数据爬虫：
import requests
import pandas as pd
import time
import re
from datetime import datetime, timedelta
class StockDataCrawler:
"""股票数据爬虫类"""
def __init__(self, symbol="sh000001"): # 默认上证指数
self.symbol = symbol
self.base_url = "http://hq.sinajs.cn/list="
self.data_file = f"data/raw/stock_{symbol}_{datetime.now().strftime('%Y%m%d')}.csv"
# 股票代码映射
self.stock_names = {
"sh000001": "上证指数",
"sz399001": "深证成指",
"sz399006": "创业板指",
"sh000300": "沪深300",
"sh000016": "上证50",
"sz399005": "中小板指"
}
def parse_sina_data(self, data_str):
"""解析新浪财经数据格式"""
# 格式: var hq_str_sh000001="上证指数,3278.1234,3280.5678,...";
if not data_str or "=" not in data_str:
return None
try:
# 提取数据部分
content = data_str.split("=")[1].strip('";')
items = content.split(",")
if len(items) < 30:
return None
# 解析各个字段
stock_data = {
"timestamp": datetime.now(),
"name": items[0], # 股票名称
"open": float(items[1]), # 今日开盘价
"close": float(items[2]), # 昨日收盘价
"price": float(items[3]), # 当前价格
"high": float(items[4]), # 今日最高价
"low": float(items[5]), # 今日最低价
"bid": float(items[6]), # 竞买价
"ask": float(items[7]), # 竞卖价
"volume": float(items[8]), # 成交量（手）
"amount": float(items[9]), # 成交额（万）
"buy1_volume": int(items[10]), # 买一量
"buy1_price": float(items[11]), # 买一价
"buy2_volume": int(items[12]),
"buy2_price": float(items[13]),
"buy3_volume": int(items[14]),
"buy3_price": float(items[15]),
"buy4_volume": int(items[16]),
"buy4_price": float(items[17]),
"buy5_volume": int(items[18]),
"buy5_price": float(items[19]),
"sell1_volume": int(items[20]), # 卖一量
"sell1_price": float(items[21]), # 卖一价
"sell2_volume": int(items[22]),
"sell2_price": float(items[23]),
"sell3_volume": int(items[24]),
"sell3_price": float(items[25]),
"sell4_volume": int(items[26]),
"sell4_price": float(items[27]),
"sell5_volume": int(items[28]),
"sell5_price": float(items[29])
}
# 计算衍生指标
stock_data["change"] = stock_data["price"] - stock_data["close"] # 涨跌
stock_data["change_percent"] = (stock_data["change"] / stock_data["close"]) * 100 # 涨跌幅
return stock_data
except Exception as e:
print(f"解析股票数据失败: {e}")
return None
def fetch_realtime_data(self):
"""获取实时股票数据"""
url = f"{self.base_url}{self.symbol}"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Referer": "http://finance.sina.com.cn",
"Accept": "*/*"
}
try:
response = requests.get(url, headers=headers, timeout=10)
response.encoding = "gbk" # 新浪使用GBK编码
if response.status_code == 200:
stock_data = self.parse_sina_data(response.text)
if stock_data:
stock_name = self.stock_names.get(self.symbol, stock_data["name"])
print(f"{stock_name}: {stock_data['price']:.2f} ({stock_data['change']:+.2f}, {stock_data['change_percent']:+.2f}%)")
return stock_data
else:
print(f"请求失败，状态码: {response.status_code}")
except requests.exceptions.RequestException as e:
print(f"网络请求失败: {e}")
except Exception as e:
print(f"获取数据失败: {e}")
return None
def collect_intraday_data(self, interval_seconds=30, duration_minutes=60):
"""收集日内交易数据"""
total_points = (duration_minutes * 60) // interval_seconds
all_data = []
stock_name = self.stock_names.get(self.symbol, self.symbol)
print(f"开始收集 {stock_name} 的日内数据")
print(f"采集间隔: {interval_seconds}秒，持续时间: {duration_minutes}分钟，总采集点: {total_points}")
print("-" * 60)
start_time = datetime.now()
end_time = start_time + timedelta(minutes=duration_minutes)
point_count = 0
while datetime.now() < end_time and point_count < total_points:
point_count += 1
current_time = datetime.now().strftime("%H:%M:%S")
print(f"[{current_time}] 第 {point_count}/{total_points} 次采集")
# 获取数据
stock_data = self.fetch_realtime_data()
if stock_data:
all_data.append(stock_data)
# 显示最新数据
latest = all_data[-1]
print(f" 价格: {latest['price']:.2f} | 涨跌: {latest['change']:+.2f}")
print(f" 成交量: {latest['volume']/10000:.1f}万手 | 成交额: {latest['amount']:.0f}万")
# 计算等待时间
if point_count < total_points:
wait_time = interval_seconds
print(f" 等待 {wait_time} 秒...")
time.sleep(wait_time)
print("-" * 60)
print(f"数据收集完成！共收集 {len(all_data)} 条记录")
# 保存数据
if all_data:
df = pd.DataFrame(all_data)
df.to_csv(self.data_file, index=False, encoding="utf-8-sig")
print(f"数据已保存到: {self.data_file}")
# 生成简单统计
self._generate_stats(df)
return df
return None
def _generate_stats(self, df):
"""生成数据统计"""
if len(df) == 0:
return
print("\n数据统计:")
print(f"时间范围: {df['timestamp'].min().strftime('%H:%M:%S')} - {df['timestamp'].max().strftime('%H:%M:%S')}")
print(f"价格区间: {df['price'].min():.2f} - {df['price'].max():.2f}")
print(f"平均价格: {df['price'].mean():.2f}")
print(f"最大涨幅: {df['change_percent'].max():.2f}%")
print(f"最大跌幅: {df['change_percent'].min():.2f}%")
print(f"总成交量: {df['volume'].sum()/10000:.1f}万手")
print(f"总成交额: {df['amount'].sum():.0f}万")
# 使用示例
if __name__ == "__main__":
# 创建爬虫实例
crawler = StockDataCrawler(symbol="sh000001") # 上证指数
# 测试单次采集
print("=== 测试单次采集 ===")
test_data = crawler.fetch_realtime_data()
if test_data:
print("测试成功！")
print(f"股票: {test_data['name']}")
print(f"当前价: {test_data['price']:.2f}")
print(f"涨跌: {test_data['change']:+.2f}")
print(f"涨跌幅: {test_data['change_percent']:+.2f}%")
# 开始日内数据收集（实际使用时取消注释）
# print("\n=== 开始日内数据收集 ===")
# df = crawler.collect_intraday_data(interval_seconds=60, duration_minutes=10)
# if df is not None:
# print(f"共采集 {len(df)} 条股票数据")
这个股票爬虫的特点：
- 实时性：可以按秒级频率采集数据，捕捉市场细微波动
- 数据丰富：不仅价格，还包括买卖盘口、成交量等深度信息
- 自动统计：采集完成后自动生成基本统计信息
- 错误恢复：网络中断后可以继续采集
3.3 数据清洗与预处理
原始数据往往有各种问题，我们需要先清洗再使用：
import pandas as pd
import numpy as np
from datetime import datetime
class DataCleaner:
"""数据清洗器"""
def __init__(self):
self.cleaning_log = []
def clean_weather_data(self, df):
"""清洗天气数据"""
if df is None or len(df) == 0:
return df
original_count = len(df)
self.cleaning_log.append(f"原始数据: {original_count} 条")
# 1. 处理缺失值
df_cleaned = df.copy()
# 数值型字段用前向填充
numeric_cols = ['temperature', 'humidity', 'high_temp', 'low_temp', 'aqi']
for col in numeric_cols:
if col in df_cleaned.columns:
missing_before = df_cleaned[col].isnull().sum()
df_cleaned[col] = df_cleaned[col].fillna(method='ffill').fillna(method='bfill')
missing_after = df_cleaned[col].isnull().sum()
if missing_before > 0:
self.cleaning_log.append(f" 列 '{col}': 填充了 {missing_before - missing_after} 个缺失值")
# 文本型字段用众数填充
text_cols = ['weather', 'wind_direction', 'wind_power']
for col in text_cols:
if col in df_cleaned.columns:
missing_before = df_cleaned[col].isnull().sum()
mode_value = df_cleaned[col].mode()[0] if not df_cleaned[col].mode().empty else "未知"
df_cleaned[col] = df_cleaned[col].fillna(mode_value)
missing_after = df_cleaned[col].isnull().sum()
if missing_before > 0:
self.cleaning_log.append(f" 列 '{col}': 用 '{mode_value}' 填充了 {missing_before - missing_after} 个缺失值")
# 2. 处理异常值（使用IQR方法）
numeric_cols_for_outlier = ['temperature', 'humidity']
for col in numeric_cols_for_outlier:
if col in df_cleaned.columns:
Q1 = df_cleaned[col].quantile(0.25)
Q3 = df_cleaned[col].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outliers = df_cleaned[(df_cleaned[col] < lower_bound) | (df_cleaned[col] > upper_bound)]
if len(outliers) > 0:
# 用边界值替换异常值
df_cleaned[col] = df_cleaned[col].clip(lower_bound, upper_bound)
self.cleaning_log.append(f" 列 '{col}': 修正了 {len(outliers)} 个异常值")
# 3. 数据标准化（可选）
if 'temperature' in df_cleaned.columns:
# 将温度标准化到0-1范围，方便后续映射
temp_min = df_cleaned['temperature'].min()
temp_max = df_cleaned['temperature'].max()
if temp_max > temp_min:
df_cleaned['temperature_normalized'] = (df_cleaned['temperature'] - temp_min) / (temp_max - temp_min)
self.cleaning_log.append(f" 温度标准化: {temp_min}°C ~ {temp_max}°C → 0 ~ 1")
if 'humidity' in df_cleaned.columns:
# 湿度已经是百分比，直接除以100
df_cleaned['humidity_normalized'] = df_cleaned['humidity'] / 100.0
# 4. 时间处理
if 'timestamp' in df_cleaned.columns:
# 确保时间列是datetime类型
df_cleaned['timestamp'] = pd.to_datetime(df_cleaned['timestamp'])
# 提取时间特征
df_cleaned['hour'] = df_cleaned['timestamp'].dt.hour
df_cleaned['day_part'] = df_cleaned['hour'].apply(self._categorize_day_part)
self.cleaning_log.append(f" 时间特征提取完成")
cleaned_count = len(df_cleaned)
self.cleaning_log.append(f"清洗后数据: {cleaned_count} 条")
self.cleaning_log.append(f"数据保留率: {cleaned_count/original_count*100:.1f}%")
return df_cleaned
def _categorize_day_part(self, hour):
"""将小时分类为时间段"""
if 5 <= hour < 10:
return "早晨"
elif 10 <= hour < 14:
return "中午"
elif 14 <= hour < 18:
return "下午"
elif 18 <= hour < 22:
return "晚上"
else:
return "深夜"
def clean_stock_data(self, df):
"""清洗股票数据"""
if df is None or len(df) == 0:
return df
original_count = len(df)
self.cleaning_log.append(f"原始股票数据: {original_count} 条")
df_cleaned = df.copy()
# 1. 处理缺失值
numeric_cols = ['price', 'volume', 'amount', 'change', 'change_percent']
for col in numeric_cols:
if col in df_cleaned.columns:
missing_before = df_cleaned[col].isnull().sum()
# 股票数据用线性插值，因为连续性较强
df_cleaned[col] = df_cleaned[col].interpolate(method='linear')
missing_after = df_cleaned[col].isnull().sum()
if missing_before > 0:
self.cleaning_log.append(f" 列 '{col}': 插值填充了 {missing_before - missing_after} 个缺失值")
# 2. 处理异常值（股票数据波动大，使用更宽松的标准）
if 'change_percent' in df_cleaned.columns:
# 涨跌幅超过20%视为异常（极端情况）
extreme_changes = df_cleaned[abs(df_cleaned['change_percent']) > 20]
if len(extreme_changes) > 0:
# 用前后平均值替换
for idx in extreme_changes.index:
if idx > 0 and idx < len(df_cleaned) - 1:
prev_val = df_cleaned.loc[idx-1, 'change_percent']
next_val = df_cleaned.loc[idx+1, 'change_percent']
df_cleaned.loc[idx, 'change_percent'] = (prev_val + next_val) / 2
self.cleaning_log.append(f" 修正了 {len(extreme_changes)} 个极端涨跌幅")
# 3. 计算技术指标
if 'price' in df_cleaned.columns:
# 简单移动平均
df_cleaned['price_ma5'] = df_cleaned['price'].rolling(window=5, min_periods=1).mean()
df_cleaned['price_ma10'] = df_cleaned['price'].rolling(window=10, min_periods=1).mean()
# 价格变化率
df_cleaned['price_change'] = df_cleaned['price'].pct_change() * 100
# 成交量变化
if 'volume' in df_cleaned.columns:
df_cleaned['volume_change'] = df_cleaned['volume'].pct_change() * 100
self.cleaning_log.append(f" 技术指标计算完成")
# 4. 标准化
if 'price' in df_cleaned.columns:
price_min = df_cleaned['price'].min()
price_max = df_cleaned['price'].max()
if price_max > price_min:
df_cleaned['price_normalized'] = (df_cleaned['price'] - price_min) / (price_max - price_min)
self.cleaning_log.append(f" 价格标准化: {price_min:.2f} ~ {price_max:.2f} → 0 ~ 1")
if 'volume' in df_cleaned.columns:
volume_min = df_cleaned['volume'].min()
volume_max = df_cleaned['volume'].max()
if volume_max > volume_min:
df_cleaned['volume_normalized'] = (df_cleaned['volume'] - volume_min) / (volume_max - volume_min)
cleaned_count = len(df_cleaned)
self.cleaning_log.append(f"清洗后股票数据: {cleaned_count} 条")
return df_cleaned
def print_cleaning_log(self):
"""打印清洗日志"""
print("数据清洗报告:")
print("=" * 50)
for log in self.cleaning_log:
print(log)
print("=" * 50)
# 使用示例
if __name__ == "__main__":
# 模拟一些天气数据
test_weather_data = {
'timestamp': pd.date_range('2024-01-01', periods=10, freq='H'),
'temperature': [15, 16, None, 18, 19, 100, 21, 22, 23, 24], # 包含缺失值和异常值
'humidity': [60, 65, 70, None, 75, 80, 85, 90, 95, 100],
'weather': ['晴', '晴', '多云', '阴', '雨', '雨', '雪', '雪', None, '雾']
}
df_weather = pd.DataFrame(test_weather_data)
print("原始天气数据:")
print(df_weather)
print("\n" + "="*50 + "\n")
# 清洗数据
cleaner = DataCleaner()
df_cleaned = cleaner.clean_weather_data(df_weather)
print("清洗后的天气数据:")
print(df_cleaned[['timestamp', 'temperature', 'humidity', 'weather', 'temperature_normalized']])
print("\n")
cleaner.print_cleaning_log()
数据清洗的关键点：
- 缺失值处理：根据数据类型选择填充方法（前向填充、众数填充、插值）
- 异常值检测：使用统计方法识别和处理异常数据
- 数据标准化：将不同尺度的数据映射到统一范围
- 特征工程：从原始数据中提取更有用的特征
4. 数据到动画：参数映射的艺术
这是整个项目的核心创意部分——如何把枯燥的数字变成生动的动画参数。不同的数据需要不同的映射策略，我分享几个实用的映射方法。
4.1 天气数据的动画映射
天气数据有很强的视觉对应关系，温度对应色调，湿度对应雾气，风力对应运动强度：
class WeatherAnimationMapper:
"""天气数据到动画参数的映射器"""
def __init__(self):
# 颜色映射：温度 -> 色温
self.temperature_colors = {
"arctic": {"rgb": [200, 230, 255], "temperature": 9000, "mood": "冰冷"},
"cold": {"rgb": [180, 220, 255], "temperature": 7000, "mood": "寒冷"},
"cool": {"rgb": [220, 240, 255], "temperature": 5000, "mood": "凉爽"},
"neutral": {"rgb": [255, 255, 255], "temperature": 4000, "mood": "舒适"},
"warm": {"rgb": [255, 240, 200], "temperature": 3000, "mood": "温暖"},
"hot": {"rgb": [255, 220, 180], "temperature": 2000, "mood": "炎热"},
"tropical": {"rgb": [255, 200, 150], "temperature": 1000, "mood": "酷热"}
}
# 天气类型映射
self.weather_styles = {
"晴": {
"lighting": "bright_sunny",
"clouds": "few_clouds",
"atmosphere": "clear",
"special_effects": ["sun_flares", "light_rays"]
},
"多云": {
"lighting": "soft_diffuse",
"clouds": "scattered_clouds",
"atmosphere": "hazy",
"special_effects": ["cloud_shadows"]
},
"阴": {
"lighting": "flat_overcast",
"clouds": "overcast",
"atmosphere": "gray",
"special_effects": ["foggy"]
},
"雨": {
"lighting": "dark_rainy",
"clouds": "rain_clouds",
"atmosphere": "wet",
"special_effects": ["rain_drops", "puddles", "wet_surfaces"]
},
"雪": {
"lighting": "soft_snowy",
"clouds": "snow_clouds",
"atmosphere": "white",
"special_effects": ["snow_flakes", "frost", "snow_covered"]
},
"雾": {
"lighting": "muted_foggy",
"clouds": "fog",
"atmosphere": "misty",
"special_effects": ["depth_fog", "atmospheric_perspective"]
}
}
# 风力映射
self.wind_effects = {
"微风": {"intensity": 0.2, "particles": 50, "movement": "gentle_breeze"},
"1级": {"intensity": 0.3, "particles": 100, "movement": "light_breeze"},
"2-3级": {"intensity": 0.5, "particles": 200, "movement": "moderate_breeze"},
"4-5级": {"intensity": 0.8, "particles": 400, "movement": "fresh_breeze"},
"6-7级": {"intensity": 1.2, "particles": 800, "movement": "strong_wind"},
"8级以上": {"intensity": 2.0, "particles": 1500, "movement": "storm_wind"}
}
def map_temperature(self, temp_celsius):
"""映射温度到视觉参数"""
if temp_celsius < -10:
color_profile = self.temperature_colors["arctic"]
animation_params = {
"color_preset": "arctic_blue",
"brightness": 0.7,
"contrast": 1.3,
"saturation": 0.8,
"hue_shift": -0.1 # 偏蓝
}
elif temp_celsius < 0:
color_profile = self.temperature_colors["cold"]
animation_params = {
"color_preset": "cold_blue",
"brightness": 0.8,
"contrast": 1.2,
"saturation": 0.9,
"hue_shift": -0.05
}
elif temp_celsius < 10:
color_profile = self.temperature_colors["cool"]
animation_params = {
"color_preset": "cool_white",
"brightness": 0.9,
"contrast": 1.1,
"saturation": 1.0,
"hue_shift": 0.0
}
elif temp_celsius < 20:
color_profile = self.temperature_colors["neutral"]
animation_params = {
"color_preset": "neutral",
"brightness": 1.0,
"contrast": 1.0,
"saturation": 1.0,
"hue_shift": 0.0
}
elif temp_celsius < 30:
color_profile = self.temperature_colors["warm"]
animation_params = {
"color_preset": "warm_yellow",
"brightness": 1.1,
"contrast": 1.1,
"saturation": 1.2,
"hue_shift": 0.05 # 偏黄
}
elif temp_celsius < 40:
color_profile = self.temperature_colors["hot"]
animation_params = {
"color_preset": "hot_orange",
"brightness": 1.2,
"contrast": 1.2,
"saturation": 1.3,
"hue_shift": 0.1 # 偏橙
}
else:
color_profile = self.temperature_colors["tropical"]
animation_params = {
"color_preset": "tropical_red",
"brightness": 1.3,
"contrast": 1.3,
"saturation": 1.4,
"hue_shift": 0.15 # 偏红
}
# 添加温度相关的动画效果
if temp_celsius > 30:
animation_params["heat_waves"] = True
animation_params["heat_distortion"] = min((temp_celsius - 30) / 10, 1.0)
elif temp_celsius < 0:
animation_params["frost_effect"] = True
animation_params["frost_intensity"] = min(abs(temp_celsius) / 10, 1.0)
return {
"visual_params": animation_params,
"color_profile": color_profile,
"description": f"{color_profile['mood']} ({temp_celsius}°C)"
}
def map_humidity(self, humidity_percent):
"""映射湿度到视觉参数"""
humidity = float(humidity_percent)
if humidity < 30:
# 干燥
return {
"fog_density": 0.1,
"atmosphere_density": 0.2,
"transparency": 0.9,
"haze": 0.1,
"particle_type": "dust",
"particle_count": int(humidity * 2)
}
elif humidity < 60:
# 舒适
return {
"fog_density": 0.3,
"atmosphere_density": 0.4,
"transparency": 0.8,
"haze": 0.3,
"particle_type": "light_mist",
"particle_count": int(humidity * 3)
}
elif humidity < 80:
# 潮湿
return {
"fog_density": 0.6,
"atmosphere_density": 0.7,
"transparency": 0.6,
"haze": 0.6,
"particle_type": "mist",
"particle_count": int(humidity * 4)
}
else:
# 非常潮湿
return {
"fog_density": 0.9,
"atmosphere_density": 0.9,
"transparency": 0.3,
"haze": 0.9,
"particle_type": "dense_fog",
"particle_count": int(humidity * 5)
}
def map_weather_type(self, weather_type):
"""映射天气类型到场景参数"""
weather_lower = str(weather_type).lower()
# 查找匹配的天气类型
for key, style in self.weather_styles.items():
if key in weather_lower:
return style
# 默认返回晴天
return self.weather_styles["晴"]
def map_wind(self, wind_description):
"""映射风力到运动参数"""
wind_lower = str(wind_description).lower()
# 查找匹配的风力描述
for key, effect in self.wind_effects.items():
if key in wind_lower:
return effect
# 默认返回微风
return self.wind_effects["微风"]
def map_time_of_day(self, hour):
"""映射时间到光照参数"""
if 5 <= hour < 8:
# 清晨
return {
"lighting_preset": "sunrise",
"light_intensity": 0.6,
"light_color": [255, 200, 150], # 暖橙色
"shadow_length": 2.0,
"shadow_softness": 0.8
}
elif 8 <= hour < 11:
# 上午
return {
"lighting_preset": "morning",
"light_intensity": 0.9,
"light_color": [255, 240, 220], # 暖白色
"shadow_length": 1.0,
"shadow_softness": 0.6
}
elif 11 <= hour < 14:
# 中午
return {
"lighting_preset": "midday",
"light_intensity": 1.2,
"light_color": [255, 255, 255], # 纯白色
"shadow_length": 0.5,
"shadow_softness": 0.4
}
elif 14 <= hour < 17:
# 下午
return {
"lighting_preset": "afternoon",
"light_intensity": 0.9,
"light_color": [255, 230, 200], # 暖黄色
"shadow_length": 1.5,
"shadow_softness": 0.7
}
elif 17 <= hour < 19:
# 黄昏
return {
"lighting_preset": "sunset",
"light_intensity": 0.7,
"light_color": [255, 150, 100], # 橙红色
"shadow_length": 2.5,
"shadow_softness": 0.9
}
else:
# 夜晚
return {
"lighting_preset": "night",
"light_intensity": 0.3,
"light_color": [100, 150, 255], # 冷蓝色
"shadow_length": 3.0,
"shadow_softness": 1.0,
"moonlight": True,
"stars": True
}
def create_animation_parameters(self, weather_data):
"""根据天气数据创建完整的动画参数"""
# 基础参数
params = {
"metadata": {
"timestamp": weather_data.get("timestamp", ""),
"location": weather_data.get("city", "未知"),
"data_source": "weather"
},
"camera": {
"movement_speed": 1.0,
"movement_type": "smooth_pan",
"fov": 60
},
"scene": {
"base_style": "realistic",
"quality": "high",
"resolution": "1024x576"
},
"animation": {
"total_frames": 16,
"fps": 8,
"loop": True
}
}
# 映射温度
if "temperature" in weather_data:
temp_params = self.map_temperature(weather_data["temperature"])
params["color_grading"] = temp_params["visual_params"]
params["metadata"]["temperature_description"] = temp_params["description"]
# 映射湿度
if "humidity" in weather_data:
humidity_params = self.map_humidity(weather_data["humidity"])
params["atmosphere"] = humidity_params
# 映射天气类型
if "weather" in weather_data:
weather_params = self.map_weather_type(weather_data["weather"])
params["weather_effects"] = weather_params
params["metadata"]["weather_type"] = weather_data["weather"]
# 映射风力
if "wind_power" in weather_data:
wind_params = self.map_wind(weather_data["wind_power"])
params["wind_effects"] = wind_params
# 映射时间
if "hour" in weather_data:
time_params = self.map_time_of_day(weather_data["hour"])
params["lighting"] = time_params
elif "timestamp" in weather_data:
# 从时间戳提取小时
try:
from datetime import datetime
ts = weather_data["timestamp"]
if isinstance(ts, str):
hour = datetime.fromisoformat(ts).hour
else:
hour = ts.hour
time_params = self.map_time_of_day(hour)
params["lighting"] = time_params
except:
pass
# 生成提示词
params["prompt"] = self.generate_prompt(weather_data)
params["negative_prompt"] = "blurry, distorted, low quality, ugly, deformed, bad anatomy"
return params
def generate_prompt(self, weather_data):
"""根据天气数据生成提示词"""
parts = []
# 时间描述
if "hour" in weather_data:
hour = weather_data["hour"]
if 5 <= hour < 8:
parts.append("early morning")
elif 8 <= hour < 11:
parts.append("morning")
elif 11 <= hour < 14:
parts.append("midday")
elif 14 <= hour < 17:
parts.append("afternoon")
elif 17 <= hour < 19:
parts.append("sunset, golden hour")
else:
parts.append("night time")
# 天气描述
if "weather" in weather_data:
weather = weather_data["weather"]
if "晴" in weather:
parts.append("sunny day")
elif "多云" in weather:
parts.append("cloudy day")
elif "阴" in weather:
parts.append("overcast day")
elif "雨" in weather:
parts.append("rainy day")
elif "雪" in weather:
parts.append("snowy day")
elif "雾" in weather:
parts.append("foggy day")
# 温度描述
if "temperature" in weather_data:
temp = weather_data["temperature"]
if temp > 30:
parts.append("hot weather")
elif temp > 20:
parts.append("warm weather")
elif temp > 10:
parts.append("cool weather")
elif temp > 0:
parts.append("cold weather")
else:
parts.append("freezing weather")
# 场景描述
if "city" in weather_data:
city = weather_data["city"]
parts.append(f"{city} cityscape")
else:
parts.append("urban cityscape")
# 质量描述
parts.append("cinematic view")
parts.append("high quality")
parts.append("8k resolution")
parts.append("photorealistic")
parts.append("detailed")
# 风格描述
parts.append("film grain")
parts.append("cinematic lighting")
parts.append("depth of field")
return ", ".join(parts)
# 使用示例
if __name__ == "__main__":
# 创建映射器
mapper = WeatherAnimationMapper()
# 测试不同天气条件的映射
test_cases = [
{"temperature": 35, "humidity": 20, "weather": "晴", "wind_power": "微风", "hour": 14},
{"temperature": 15, "humidity": 60, "weather": "多云", "wind_power": "2-3级", "hour": 9},
{"temperature": 5, "humidity": 85, "weather": "雨", "wind_power": "4-5级", "hour": 18},
{"temperature": -5, "humidity": 90, "weather": "雪", "wind_power": "1级", "hour": 22}
]
print("天气数据到动画参数映射测试:")
print("=" * 80)
for i, weather in enumerate(test_cases, 1):
print(f"\n测试案例 {i}:")
print(f" 天气条件: {weather['temperature']}°C, {weather['humidity']}%湿度, {weather['weather']}, {weather['wind_power']}, {weather['hour']}:00")
params = mapper.create_animation_parameters(weather)
print(f" 生成的提示词: {params['prompt'][:80]}...")
print(f" 颜色预设: {params['color_grading']['color_preset']}")
print(f" 天气效果: {params['weather_effects']['lighting']}")
print(f" 风力强度: {params['wind_effects']['intensity']}")
print(f" 光照预设: {params['lighting']['lighting_preset']}")
这个映射器的核心思想是：把数据特征转化为视觉特征。温度决定色调冷暖，湿度决定雾气浓度，天气类型决定场景风格，风力决定运动强度，时间决定光照角度。
4.2 股票数据的动画映射
股票数据的波动性更强，适合做更动态、更有冲击力的映射：
class StockAnimationMapper:
"""股票数据到动画参数的映射器"""
def __init__(self):
# 价格变化映射
self.price_change_mapping = {
"extreme_up": {"threshold": 5.0, "color": [0, 255, 0], "intensity": 2.0, "mood": "狂热上涨"},
"strong_up": {"threshold": 2.0, "color": [100, 255, 100], "intensity": 1.5, "mood": "强势上涨"},
"moderate_up": {"threshold": 0.5, "color": [150, 255, 150], "intensity": 1.0, "mood": "温和上涨"},
"stable": {"threshold": 0.0, "color": [200, 200, 200], "intensity": 0.5, "mood": "横盘整理"},
"moderate_down": {"threshold": -0.5, "color": [255, 150, 150], "intensity": 1.0, "mood": "温和下跌"},
"strong_down": {"threshold": -2.0, "color": [255, 100, 100], "intensity": 1.5, "mood": "强势下跌"},
"extreme_down": {"threshold": -5.0, "color": [255, 0, 0], "intensity": 2.0, "mood": "恐慌下跌"}
}
# 成交量映射
self.volume_mapping = {
"extremely_low": {"ratio": 0.3, "particles": 100, "size": 0.5, "speed": 0.3},
"low": {"ratio": 0.6, "particles": 300, "size": 0.8, "speed": 0.5},
"normal": {"ratio": 1.0, "particles": 500, "size": 1.0, "speed": 1.0},
"high": {"ratio": 2.0, "particles": 1000, "size": 1.5, "speed": 1.5},
"extremely_high": {"ratio": 5.0, "particles": 2000, "size": 2.0, "speed": 2.0}
}
# 市场情绪映射
self.market_sentiment = {
"bullish": {
"camera_movement": "rising",
"particle_direction": "upward",
"lighting": "bright_optimistic",
"music_tempo": "fast_upbeat"
},
"bearish": {
"camera_movement": "falling",
"particle_direction": "downward",
"lighting": "dark_gloomy",
"music_tempo": "slow_dark"
},
"neutral": {
"camera_movement": "floating",
"particle_direction": "swirling",
"lighting": "neutral_calm",
"music_tempo": "medium_steady"
}
}
def map_price_change(self, change_percent, current_price, avg_price):
"""映射价格变化到视觉参数"""
# 确定变化级别
change_level = "stable"
for level, config in self.price_change_mapping.items():
if change_percent >= config["threshold"]:
change_level = level
break
config = self.price_change_mapping[change_level]
# 计算相对强度
price_ratio = current_price / avg_price if avg_price > 0 else 1.0
intensity_multiplier = min(abs(price_ratio - 1.0) * 2 + 1.0, 3.0)
# 构建视觉参数
visual_params = {
"color_scheme": {
"primary_color": config["color"],
"secondary_color": self._adjust_color(config["color"], -30),
"intensity": config["intensity"] * intensity_multiplier,
"bloom_effect": min(config["intensity"] * 0.5, 1.0)
},
"movement": {
"direction": "up" if change_percent > 0 else "down",
"speed": min(abs(change_percent) * 0.1, 2.0),
"amplitude": min(abs(change_percent) * 0.05, 1.0),
"frequency": min(abs(change_percent) * 0.2, 3.0)
},
"particles": {
"enabled": True,
"count": int(abs(change_percent) * 50),
"size": min(abs(change_percent) * 0.1, 2.0),
"lifetime": min(abs(change_percent) * 0.5, 5.0)
}
}
# 添加特殊效果
if abs(change_percent) > 3.0:
visual_params["special_effects"] = ["energy_pulses", "light_trails"]
if abs(change_percent) > 7.0:
visual_params["special_effects"].append("shockwaves")
return {
"visual_params": visual_params,
"change_level": change_level,
"mood": config["mood"],
"intensity": config["intensity"] * intensity_multiplier
}
def map_volume(self, current_volume, avg_volume):
"""映射成交量到视觉参数"""
if avg_volume == 0:
volume_ratio = 1.0
else:
volume_ratio = current_volume / avg_volume
# 确定成交量级别
volume_level = "normal"
for level, config in self.volume_mapping.items():
if volume_ratio <= config["ratio"]:
volume_level = level
break
config = self.volume_mapping[volume_level]
# 调整参数基于实际比例
adjusted_particles = int(config["particles"] * min(volume_ratio, 3.0))
adjusted_size = config["size"] * min(volume_ratio, 2.0)
adjusted_speed = config["speed"] * min(volume_ratio, 2.0)
return {
"particle_system": {
"count": adjusted_particles,
"size": adjusted_size,
"speed": adjusted_speed,
"density": min(volume_ratio, 2.0),
"chaos": min(volume_ratio * 0.5, 1.0)
},
"volume_level": volume_level,
"volume_ratio": volume_ratio
}
def map_market_sentiment(self, price_trend, volume_trend, volatility):
"""映射市场情绪到场景参数"""
# 简单情绪判断逻辑
if price_trend > 0.1 and volume_trend > 0:
sentiment = "bullish"
elif price_trend < -0.1 and volume_trend > 0:
sentiment = "bearish"
else:
sentiment = "neutral"
config = self.market_sentiment[sentiment]
# 根据波动性调整强度
volatility_factor = min(volatility * 2, 2.0)
scene_params = {
"camera": {
"movement_type": config["camera_movement"],
"movement_speed": 1.0 * volatility_factor,
"shake_intensity": min(volatility * 0.5, 1.0)
},
"particles": {
"direction": config["particle_direction"],
"spread": volatility * 0.5,
"turbulence": volatility
},
"lighting": {
"preset": config["lighting"],
"intensity": 1.0 + volatility * 0.5,
"contrast": 1.0 + volatility * 0.3
},
"audio": {
"tempo": config["music_tempo"],
"intensity": volatility_factor,
"pitch_variation": volatility * 0.1
}
}
# 高波动性添加特效
if volatility > 0.5:
scene_params["special_effects"] = ["glitch_effects", "color_shift"]
if volatility > 1.0:
scene_params["special_effects"].append("screen_shake")
return {
"scene_params": scene_params,
"sentiment": sentiment,
"volatility_factor": volatility_factor
}
def _adjust_color(self, rgb_color, adjustment):
"""调整颜色亮度"""
return [max(0, min(255, c + adjustment)) for c in rgb_color]
def create_animation_sequence(self, stock_data_frame):
"""根据股票数据序列创建动画序列"""
if stock_data_frame is None or len(stock_data_frame) == 0:
return []
# 计算技术指标
df = stock_data_frame.copy()
# 移动平均线
if 'price' in df.columns:
df['price_ma5'] = df['price'].rolling(window=5, min_periods=1).mean()
df['price_ma20'] = df['price'].rolling(window=20, min_periods=1).mean()
# 成交量平均
if 'volume' in df.columns:
df['volume_ma5'] = df['volume'].rolling(window=5, min_periods=1).mean()
# 波动率
if 'price' in df.columns:
df['returns'] = df['price'].pct_change()
df['volatility'] = df['returns'].rolling(window=10, min_periods=1).std()
animation_sequence = []
for i, row in df.iterrows():
frame_params = {
"frame_index": len(animation_sequence),
"timestamp": row.get("timestamp", i),
"data_point": {
"price": row.get("price", 0),
"volume": row.get("volume", 0),
"change_percent": row.get("change_percent", 0)
},
"animation_params": {}
}
# 映射价格变化
if 'price' in row and 'price_ma20' in row:
price_change_params = self.map_price_change(
row.get('change_percent', 0),
row['price'],
row['price_ma20']
)
frame_params["animation_params"]["price_effects"] = price_change_params["visual_params"]
frame_params["mood"] = price_change_params["mood"]
# 映射成交量
if 'volume' in row and 'volume_ma5' in row:
volume_params = self.map_volume(
row['volume'],
row['volume_ma5']
)
frame_params["animation_params"]["volume_effects"] = volume_params["particle_system"]
frame_params["volume_level"] = volume_params["volume_level"]
# 映射市场情绪
if 'price_ma5' in row and 'volume_ma5' in row and 'volatility' in row:
price_trend = 1 if row['price'] > row['price_ma5'] else -1
volume_trend = 1 if row['volume'] > row['volume_ma5'] else -1
sentiment_params = self.map_market_sentiment(
price_trend,
volume_trend,
row.get('volatility', 0)
)
frame_params["animation_params"]["scene_settings"] = sentiment_params["scene_params"]
frame_params["sentiment"] = sentiment_params["sentiment"]
# 生成提示词
frame_params["prompt"] = self.generate_stock_prompt(row)
frame_params["negative_prompt"] = "blurry, distorted, low quality, ugly, simple, boring"
animation_sequence.append(frame_params)
return animation_sequence
def generate_stock_prompt(self, stock_data):
"""生成股票数据提示词"""
parts = []
# 根据价格变化确定基调
change_percent = stock_data.get('change_percent', 0)
if change_percent > 5:
parts.append("explosive growth")
parts.append("bull market rally")
parts.append("green energy flowing")
elif change_percent > 2:
parts.append("strong upward trend")
parts.append("optimistic trading")
parts.append("rising momentum")
elif change_percent > 0:
parts.append("moderate gains")
parts.append("steady growth")
parts.append("positive movement")
elif change_percent > -2:
parts.append("sideways trading")
parts.append("market consolidation")
parts.append("balanced flow")
elif change_percent > -5:
parts.append("downward pressure")
parts.append("bearish sentiment")
parts.append("red waves crashing")
else:
parts.append("market crash")
parts.append("panic selling")
parts.append("chaotic downturn")
# 根据成交量添加描述
volume = stock_data.get('volume', 0)
if volume > 10000000:
parts.append("high volume trading")
parts.append("intense activity")
elif volume > 1000000:
parts.append("moderate volume")
parts.append("active market")
else:
parts.append("low volume")
parts.append("quiet trading")
# 添加抽象金融可视化描述
parts.append("abstract financial visualization")
parts.append("data flow animation")
parts.append("digital particles")
parts.append("market energy")
# 添加质量描述
parts.append("cinematic")
parts.append("high quality")
parts.append("8k resolution")
parts.append("detailed")
parts.append("dynamic")
# 添加风格描述
parts.append("futuristic")
parts.append("holographic")
parts.append("neon glow")
parts.append("cyberpunk aesthetic")
return ", ".join(parts)
# 使用示例
if __name__ == "__main__":
# 创建映射器
mapper = StockAnimationMapper()
# 创建测试数据
import pandas as pd
import numpy as np
# 生成模拟股票数据
dates = pd.date_range('2024-01-01 09:30', periods=20, freq='5min')
prices = 100 + np.cumsum(np.random.randn(20) * 0.5)
volumes = np.random.randint(1000000, 10000000, 20)
test_data = pd.DataFrame({
'timestamp': dates,
'price': prices,
'volume': volumes,
'change_percent': np.random.randn(20) * 2
})
# 计算移动平均
test_data['price_ma20'] = test_data['price'].rolling(window=5, min_periods=1).mean()
test_data['volume_ma5'] = test_data['volume'].rolling(window=5, min_periods=1).mean()
test_data['returns'] = test_data['price'].pct_change()
test_data['volatility'] = test_data['returns'].rolling(window=5, min_periods=1).std()
print("股票数据到动画参数映射测试:")
print("=" * 80)
print(f"数据点数: {len(test_data)}")
print(f"价格范围: {test_data['price'].min():.2f} - {test_data['price'].max():.2f}")
print(f"涨跌幅范围: {test_data['change_percent'].min():.2f}% - {test_data['change_percent'].max():.2f}%")
print("\n" + "=" * 80)
# 生成动画序列
animation_sequence = mapper.create_animation_sequence(test_data)
print(f"生成的动画序列: {len(animation_sequence)} 帧")
# 显示前3帧的参数
for i in range(min(3, len(animation_sequence))):
frame = animation_sequence[i]
print(f"\n第 {i+1} 帧:")
print(f" 时间: {frame['timestamp']}")
print(f" 价格: {frame['data_point']['price']:.2f}")
print(f" 涨跌幅: {frame['data_point']['change_percent']:.2f}%")
print更多推荐
2
0- 0
已为社区贡献29条内容
所有评论(0)