45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
from abc import ABC, abstractmethod
|
|
import pandas as pd
|
|
from typing import Dict, List
|
|
|
|
class BaseStrategy(ABC):
|
|
|
|
def __init__(self, name: str, description: str):
|
|
self.name = name
|
|
self.description = description
|
|
self.weights = {}
|
|
|
|
@abstractmethod
|
|
def calculate_score(self, stock_data: Dict) -> float:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_criteria(self) -> Dict:
|
|
pass
|
|
|
|
def filter_stocks(self, stocks_data: List[Dict], min_score: float = 60) -> List[Dict]:
|
|
filtered_stocks = []
|
|
|
|
for stock in stocks_data:
|
|
try:
|
|
score = self.calculate_score(stock)
|
|
if score >= min_score:
|
|
stock['strategy_score'] = score
|
|
stock['strategy_name'] = self.name
|
|
filtered_stocks.append(stock)
|
|
except Exception as e:
|
|
print(f"Error calculating score for {stock.get('ts_code', 'unknown')}: {e}")
|
|
continue
|
|
|
|
return sorted(filtered_stocks, key=lambda x: x['strategy_score'], reverse=True)
|
|
|
|
def rank_stocks(self, stocks_data: List[Dict]) -> List[Dict]:
|
|
for stock in stocks_data:
|
|
try:
|
|
stock['strategy_score'] = self.calculate_score(stock)
|
|
stock['strategy_name'] = self.name
|
|
except Exception as e:
|
|
print(f"Error calculating score for {stock.get('ts_code', 'unknown')}: {e}")
|
|
stock['strategy_score'] = 0
|
|
|
|
return sorted(stocks_data, key=lambda x: x['strategy_score'], reverse=True) |