FilesExpand file tree

master

/

debug_factor_calculation.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

84 lines (72 loc) · 3.34 KB

master

/

debug_factor_calculation.py

Copy path

Top

File metadata and controls

• Code
• Blame

84 lines (72 loc) · 3.34 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

#!/usr/bin/env python

# -*- coding: utf-8 -*-

"""

调试因子计算过程

"""

from app import create_app

from app.extensions import db

from app.services.factor_engine import FactorEngine

import pandas as pd

def debug_factor_calculation():

app = create_app()

with app.app_context():

try:

# 创建因子引擎

factor_engine = FactorEngine()

# 测试数据获取

print("=== 测试数据获取 ===")

trade_date = "2025-05-23"

# 获取少量股票进行测试

with db.engine.connect() as conn:

result = conn.execute(db.text('SELECT DISTINCT ts_code FROM stock_moneyflow LIMIT 5'))

test_stocks = [row[0] for row in result.fetchall()]

print(f"测试股票: {test_stocks}")

# 测试获取资金流向数据

data = factor_engine._get_factor_data('money_flow_strength', test_stocks, trade_date, trade_date)

print(f"获取到的数据类型: {list(data.keys())}")

if 'moneyflow' in data:

moneyflow_df = data['moneyflow']

print(f"资金流向数据形状: {moneyflow_df.shape}")

print(f"资金流向数据列: {list(moneyflow_df.columns)}")

print(f"资金流向数据前5行:")

print(moneyflow_df.head())

# 检查是否有空值

print(f"空值情况:")

print(moneyflow_df.isnull().sum())

# 测试计算资金流向强度因子

print("\n=== 测试资金流向强度因子计算 ===")

result = factor_engine._money_flow_strength_factor(data, 'money_flow_strength')

print(f"计算结果形状: {result.shape}")

if not result.empty:

print("计算结果前5行:")

print(result.head())

else:

print("计算结果为空")

else:

print("未获取到资金流向数据")

# 测试筹码数据

print("\n=== 测试筹码数据 ===")

chip_data = factor_engine._get_factor_data('chip_concentration', test_stocks, trade_date, trade_date)

if 'cyq' in chip_data:

cyq_df = chip_data['cyq']

print(f"筹码数据形状: {cyq_df.shape}")

print(f"筹码数据列: {list(cyq_df.columns)}")

print(f"筹码数据前5行:")

print(cyq_df.head())

# 测试计算筹码集中度因子

print("\n=== 测试筹码集中度因子计算 ===")

result = factor_engine._chip_concentration_factor(chip_data, 'chip_concentration')

print(f"计算结果形状: {result.shape}")

if not result.empty:

print("计算结果前5行:")

print(result.head())

else:

print("计算结果为空")

else:

print("未获取到筹码数据")

except Exception as e:

print(f"调试失败: {e}")

import traceback

traceback.print_exc()

if __name__ == '__main__':

debug_factor_calculation()