Spaces:
Running
Running
File size: 4,046 Bytes
5c793cc | 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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | import asyncio
from collections import Counter
from math import sqrt
from app.modules.posts.domain.clustering import ClusterRunResult, KMeansResult
from app.modules.posts.domain.ports.cluster_training_repository import ClusterTrainingRepository
from app.modules.posts.domain.ports.clusterer import PostClusterer
class TrainPostClustersUseCase:
ALGORITHM_VERSION = "sklearn-minibatch-kmeans-v1"
def __init__(
self,
repository: ClusterTrainingRepository,
clusterer: PostClusterer,
min_posts: int = 20,
min_k: int = 2,
max_k: int = 50,
lookback_days: int = 90,
min_cluster_size: int = 2,
max_cluster_ratio: float = 0.70,
auto_activate: bool = False,
) -> None:
self._repository = repository
self._clusterer = clusterer
self._min_posts = min_posts
self._min_k = min_k
self._max_k = max_k
self._lookback_days = lookback_days
self._min_cluster_size = min_cluster_size
self._max_cluster_ratio = max_cluster_ratio
self._auto_activate = auto_activate
async def execute(self) -> ClusterRunResult:
embeddings = await self._repository.list_active_embeddings(self._lookback_days)
if len(embeddings) < self._min_posts:
raise ValueError(
f"se requieren al menos {self._min_posts} posts para entrenar K-Means"
)
estimated = round(sqrt(len(embeddings) / 2))
upper = min(self._max_k, len(embeddings) - 1)
lower = min(self._min_k, upper)
candidate_ks = sorted(
{
max(lower, min(upper, value))
for value in (estimated - 4, estimated, estimated + 4)
}
)
run_id = await self._repository.create_run(
planned_k=candidate_ks[len(candidate_ks) // 2],
sample_size=len(embeddings),
algorithm_version=self.ALGORITHM_VERSION,
parameters={
"candidate_ks": ",".join(str(value) for value in candidate_ks),
"min_cluster_size": self._min_cluster_size,
"max_cluster_ratio": self._max_cluster_ratio,
},
)
try:
evaluated: list[tuple[int, KMeansResult]] = []
for k in candidate_ks:
result = await asyncio.to_thread(self._clusterer.fit, embeddings, k)
if _is_valid_distribution(
result,
len(embeddings),
self._min_cluster_size,
self._max_cluster_ratio,
):
evaluated.append((k, result))
if not evaluated:
raise ValueError("ningun K produjo una distribucion de clusters valida")
k, result = max(
evaluated,
key=lambda item: (
item[1].silhouette_score
if item[1].silhouette_score is not None
else float("-inf"),
-item[1].inertia,
),
)
await self._repository.save_validated_run(run_id, k, result)
status = "validated"
if self._auto_activate:
await self._repository.activate_run(run_id)
status = "active"
return ClusterRunResult(
run_id,
k,
len(embeddings),
result.inertia,
result.silhouette_score,
status,
)
except Exception as exc:
await self._repository.mark_run_failed(run_id, str(exc))
raise
def _is_valid_distribution(
result: KMeansResult,
sample_size: int,
min_cluster_size: int,
max_cluster_ratio: float,
) -> bool:
counts = Counter(item.cluster_id for item in result.assignments)
if not counts or min(counts.values()) < min_cluster_size:
return False
return max(counts.values()) / sample_size <= max_cluster_ratio
|